Add Subscription API to Indexer (#599)

* add subscribe API

* workaround v6

* refactor subscriptions APIs

* delete indexer permisssions

* permissions.go: revert

* hotfix service_test.go

* Simplify

* Add topics (scripts) to GetSubscriptionResponse message

* Fix shutdown and logging

* Fixes

---------

Co-authored-by: altafan <[email protected]>

Author: louisinger

api-spec/openapi/swagger/ark/v1/indexer.swagger.json

@@ -478,6 +478,109 @@
         ]
       }
     },
+    "/v1/script/subscribe": {
+      "post": {
+        "operationId": "IndexerService_SubscribeForScripts",
+        "responses": {
+          "200": {
+            "description": "A successful response.",
+            "schema": {
+              "$ref": "#/definitions/v1SubscribeForScriptsResponse"
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/rpcStatus"
+            }
+          }
+        },
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/v1SubscribeForScriptsRequest"
+            }
+          }
+        ],
+        "tags": [
+          "IndexerService"
+        ]
+      }
+    },
+    "/v1/script/subscription/{subscriptionId}": {
+      "get": {
+        "operationId": "IndexerService_GetSubscription",
+        "responses": {
+          "200": {
+            "description": "A successful response.(streaming responses)",
+            "schema": {
+              "type": "object",
+              "properties": {
+                "result": {
+                  "$ref": "#/definitions/v1GetSubscriptionResponse"
+                },
+                "error": {
+                  "$ref": "#/definitions/rpcStatus"
+                }
+              },
+              "title": "Stream result of v1GetSubscriptionResponse"
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/rpcStatus"
+            }
+          }
+        },
+        "parameters": [
+          {
+            "name": "subscriptionId",
+            "in": "path",
+            "required": true,
+            "type": "string"
+          }
+        ],
+        "tags": [
+          "IndexerService"
+        ]
+      }
+    },
+    "/v1/script/unsubscribe": {
+      "post": {
+        "operationId": "IndexerService_UnsubscribeForScripts",
+        "responses": {
+          "200": {
+            "description": "A successful response.",
+            "schema": {
+              "$ref": "#/definitions/v1UnsubscribeForScriptsResponse"
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/rpcStatus"
+            }
+          }
+        },
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/v1UnsubscribeForScriptsRequest"
+            }
+          }
+        ],
+        "tags": [
+          "IndexerService"
+        ]
+      }
+    },
     "/v1/virtualTx/{txids}": {
       "get": {
         "operationId": "IndexerService_GetVirtualTxs",
@@ -687,6 +790,34 @@
         }
       }
     },
+    "v1GetSubscriptionResponse": {
+      "type": "object",
+      "properties": {
+        "txid": {
+          "type": "string"
+        },
+        "scripts": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
+        "newVtxos": {
+          "type": "array",
+          "items": {
+            "type": "object",
+            "$ref": "#/definitions/v1IndexerVtxo"
+          }
+        },
+        "spentVtxos": {
+          "type": "array",
+          "items": {
+            "type": "object",
+            "$ref": "#/definitions/v1IndexerVtxo"
+          }
+        }
+      }
+    },
     "v1GetSweptCommitmentTxResponse": {
       "type": "object",
       "properties": {
@@ -1003,6 +1134,47 @@
           "type": "string"
         }
       }
+    },
+    "v1SubscribeForScriptsRequest": {
+      "type": "object",
+      "properties": {
+        "scripts": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
+        "subscriptionId": {
+          "type": "string",
+          "title": "if set, update an existing subscription"
+        }
+      }
+    },
+    "v1SubscribeForScriptsResponse": {
+      "type": "object",
+      "properties": {
+        "subscriptionId": {
+          "type": "string"
+        }
+      }
+    },
+    "v1UnsubscribeForScriptsRequest": {
+      "type": "object",
+      "properties": {
+        "subscriptionId": {
+          "type": "string"
+        },
+        "scripts": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          },
+          "title": "if empty, unsubscribe all scripts"
+        }
+      }
+    },
+    "v1UnsubscribeForScriptsResponse": {
+      "type": "object"
     }
   }
 }

api-spec/protobuf/ark/v1/indexer.proto

@@ -65,6 +65,23 @@ service IndexerService {
       get: "/v1/commitmentTx/{txid}/swept"
     };
   }
+  rpc SubscribeForScripts(SubscribeForScriptsRequest) returns (SubscribeForScriptsResponse) {
+    option (google.api.http) = {
+      post: "/v1/script/subscribe"
+      body: "*"
+    };
+  };
+  rpc UnsubscribeForScripts(UnsubscribeForScriptsRequest) returns (UnsubscribeForScriptsResponse) {
+    option (google.api.http) = {
+      post: "/v1/script/unsubscribe"
+      body: "*"
+    };
+  };
+  rpc GetSubscription(GetSubscriptionRequest) returns (stream GetSubscriptionResponse) {
+    option (google.api.http) = {
+      get: "/v1/script/subscription/{subscription_id}"
+    };
+  };
 }
 
 message GetCommitmentTxRequest {
@@ -183,15 +200,6 @@ message GetSweptCommitmentTxResponse {
   repeated string swept_by = 1;
 }
 
-message SubscribeForAddressesRequest {
-  repeated string addresses = 1;
-}
-message SubscribeForAddressesResponse {
-  string address = 1;
-  repeated IndexerVtxo new_vtxos = 2;
-  repeated IndexerVtxo spent_vtxos = 3;
-}
-
 message IndexerBatch {
   uint64 total_output_amount = 1;
   int32 total_output_vtxos = 2;
@@ -268,4 +276,33 @@ message IndexerPageResponse {
   int32 current = 1;
   int32 next = 2;
   int32 total = 3;
+}
+
+message SubscribeForScriptsRequest {
+  repeated string scripts = 1;
+  // if set, update an existing subscription
+  string subscription_id = 2;
+}
+
+message SubscribeForScriptsResponse {
+  string subscription_id = 1;
+}
+
+message UnsubscribeForScriptsRequest {
+  string subscription_id = 1;
+  // if empty, unsubscribe all scripts
+  repeated string scripts = 2; 
+}
+
+message UnsubscribeForScriptsResponse {}
+
+message GetSubscriptionRequest {
+  string subscription_id = 1;
+}
+
+message GetSubscriptionResponse {
+  string txid = 1;
+  repeated string scripts = 2;
+  repeated IndexerVtxo new_vtxos = 3;
+  repeated IndexerVtxo spent_vtxos = 4;
 }