feat(root): proxy source's MQTT resync for system admins

OPENAPI_DOC.yml

@@ -26514,29 +26514,20 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/PlaceOS.Api.Application.ContentError'
-  /api/engine/v2/reindex:
-    post:
-      summary: 'DEPRECATED: search is served directly from PostgreSQL (PPT-2644);
-        there'
-      description: 'DEPRECATED: search is served directly from PostgreSQL (PPT-2644);
-        there
-
-        is no external index to rebuild. Kept as a no-op so existing admin UIs
-
-        and automation don''t error. Remove in a future release.'
+  /api/engine/v2/source/resync:
+    get:
+      summary: where the last resync of the MQTT brokers got to
       tags:
       - Root
-      operationId: PlaceOS::Api::Root_reindex
-      parameters:
-      - name: backfill
-        in: query
-        description: deprecated, has no effect
-        example: true
-        schema:
-          type: boolean
+      operationId: PlaceOS::Api::Root_source_resync_status
+      parameters: []
       responses:
         "200":
           description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PlaceOS.Api.Root.SourceResyncStatus'
         "409":
           description: Conflict
           content:
@@ -26615,23 +26606,36 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/PlaceOS.Api.Application.ContentError'
-  /api/engine/v2/backfill:
     post:
-      summary: 'DEPRECATED: search is served directly from PostgreSQL (PPT-2644);
-        the'
-      description: 'DEPRECATED: search is served directly from PostgreSQL (PPT-2644);
-        the
+      summary: publishes every module's state and every model's metadata to the MQTT
+        brokers again.
+      description: 'publishes every module''s state and every model''s metadata to
+        the MQTT brokers again.
 
-        database is always authoritative. Kept as a no-op so existing admin UIs
+        With `clear`, everything source publishes is removed from the brokers first,
+        for when
 
-        and automation don''t error. Remove in a future release.'
+        what they hold has drifted. Runs in the background: 202 once started, 409
+        while one is
+
+        running. Progress is at `GET /source/resync`'
       tags:
       - Root
-      operationId: PlaceOS::Api::Root_backfill
-      parameters: []
+      operationId: PlaceOS::Api::Root_source_resync
+      parameters:
+      - name: clear
+        in: query
+        description: remove everything source publishes from the brokers first
+        example: true
+        schema:
+          type: boolean
       responses:
-        "200":
-          description: OK
+        "202":
+          description: Accepted
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PlaceOS.Api.Root.SourceResyncStatus'
         "409":
           description: Conflict
           content:
@@ -46004,6 +46008,39 @@ components:
       - version
       - build_time
       - platform_version
+    PlaceOS.Api.Root.SourceResyncStatus:
+      type: object
+      properties:
+        state:
+          type: string
+        clear:
+          type: boolean
+        started_at:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: "null"
+        finished_at:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: "null"
+        cleared:
+          type: integer
+          format: Int64
+        metadata:
+          type: integer
+          format: Int32
+        error:
+          anyOf:
+          - type: "null"
+          - type: string
+      required:
+      - state
+      - clear
+      - cleared
+      - metadata
+      description: Where a resync of the MQTT brokers got to, as source reports it
     PlaceOS.Model.JsonSchema:
       type: object
       properties:

spec/controllers/root_spec.cr

@@ -51,6 +51,52 @@ module PlaceOS::Api
       end
     end
 
+    describe "source resync" do
+      status_json = ->(state : String, clear : Bool) {
+        {state: state, clear: clear, started_at: Time.utc, finished_at: nil, cleared: 0, metadata: 0, error: nil}.to_json
+      }
+
+      it "starts a resync with clearing, for a system admin" do
+        HttpMocks.reset
+        WebMock.stub(:post, /\/api\/source\/v1\/resync\?clear=true$/)
+          .to_return(status: 202, body: status_json.call("clearing", true))
+
+        result = client.post(File.join(Root.base_route, "source/resync?clear=true"), headers: Spec::Authentication.headers)
+        result.status_code.should eq 202
+        status = Root::SourceResyncStatus.from_json(result.body)
+        status.state.should eq "clearing"
+        status.clear?.should be_true
+      end
+
+      it "reports one already running" do
+        HttpMocks.reset
+        WebMock.stub(:post, /\/api\/source\/v1\/resync\?clear=false$/)
+          .to_return(status: 409, body: status_json.call("resyncing", false))
+
+        result = client.post(File.join(Root.base_route, "source/resync"), headers: Spec::Authentication.headers)
+        result.status_code.should eq 409
+        Root::SourceResyncStatus.from_json(result.body).state.should eq "resyncing"
+      end
+
+      it "reports where the last one got to" do
+        HttpMocks.reset
+        WebMock.stub(:get, /\/api\/source\/v1\/resync$/)
+          .to_return(status: 200, body: status_json.call("complete", true))
+
+        result = client.get(File.join(Root.base_route, "source/resync"), headers: Spec::Authentication.headers)
+        result.status_code.should eq 200
+        Root::SourceResyncStatus.from_json(result.body).state.should eq "complete"
+      end
+
+      it "is for system admins only" do
+        HttpMocks.reset
+        headers = Spec::Authentication.headers(sys_admin: false, support: true)
+
+        client.post(File.join(Root.base_route, "source/resync"), headers: headers).status_code.should eq 403
+        client.get(File.join(Root.base_route, "source/resync"), headers: headers).status_code.should eq 403
+      end
+    end
+
     describe "POST /signal" do
       it "writes an arbitrary payload to a redis subscription on scoped channel" do
         user, headers = Spec::Authentication.authenticated

src/placeos-rest-api/controllers/root.cr

@@ -212,21 +212,52 @@ module PlaceOS::Api
       end
     end
 
-    # DEPRECATED: search is served directly from PostgreSQL (PPT-2644); there
-    # is no external index to rebuild. Kept as a no-op so existing admin UIs
-    # and automation don't error. Remove in a future release.
-    @[AC::Route::POST("/reindex")]
-    def reindex(
-      @[AC::Param::Info(description: "deprecated, has no effect", example: "true")]
-      backfill : Bool = false,
-    ) : Nil
+    # MQTT
+    ###############################################################################################
+
+    # Where a resync of the MQTT brokers got to, as source reports it
+    struct SourceResyncStatus
+      include JSON::Serializable
+
+      # idle, clearing, resyncing, complete or failed
+      getter state : String
+      # whether everything source publishes was removed from the brokers first
+      getter? clear : Bool
+      getter started_at : Time?
+      getter finished_at : Time?
+      # retained topics removed, when clearing
+      getter cleared : Int64
+      # models whose metadata was published again
+      getter metadata : Int32
+      getter error : String?
     end
 
-    # DEPRECATED: search is served directly from PostgreSQL (PPT-2644); the
-    # database is always authoritative. Kept as a no-op so existing admin UIs
-    # and automation don't error. Remove in a future release.
-    @[AC::Route::POST("/backfill")]
-    def backfill : Nil
+    # publishes every module's state and every model's metadata to the MQTT brokers again.
+    # With `clear`, everything source publishes is removed from the brokers first, for when
+    # what they hold has drifted. Runs in the background: 202 once started, 409 while one is
+    # running. Progress is at `GET /source/resync`
+    @[AC::Route::POST("/source/resync", status_code: HTTP::Status::ACCEPTED)]
+    def source_resync(
+      @[AC::Param::Info(description: "remove everything source publishes from the brokers first", example: "true")]
+      clear : Bool = false,
+    ) : SourceResyncStatus
+      response = Root.source_request("POST", "/api/source/v1/resync?clear=#{clear}")
+      status = SourceResyncStatus.from_json(response.body)
+      render status: :conflict, json: status if response.status_code == 409
+      status
+    end
+
+    # where the last resync of the MQTT brokers got to
+    @[AC::Route::GET("/source/resync")]
+    def source_resync_status : SourceResyncStatus
+      SourceResyncStatus.from_json(Root.source_request("GET", "/api/source/v1/resync").body)
+    end
+
+    protected def self.source_request(method : String, path : String) : HTTP::Client::Response
+      uri = URI.new(host: PLACE_SOURCE_HOST, port: PLACE_SOURCE_PORT, scheme: "http")
+      response = HTTP::Client.exec(method, "#{uri}#{path}")
+      return response if response.success? || response.status_code == 409
+      raise "source responded to #{method} #{path} with #{response.status_code}"
     end
   end
 end