Merge e98c03e702efaed305be0d1a5379ea1b3a8d2f2f into 5c0bd056a73576fcd6ac4db36c9fdb6e070af616

Add swagger definitions for the receive end point

.github/workflows/check-docs.yml

@@ -0,0 +1,28 @@
+name: Check Generated Docs
+
+on:
+  workflow_call:
+
+permissions:
+  contents: read
+
+jobs:
+  check-docs:
+    runs-on: ubuntu-24.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Set up Go
+        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6.0.0
+        with:
+          go-version-file: src/go.mod
+          cache-dependency-path: src/go.sum
+
+      - name: Regenerate docs
+        working-directory: src
+        run: go run github.com/swaggo/swag/cmd/swag@v1.16.6 init --requiredByDefault
+
+      - name: Fail if docs are out of date
+        run: |
+          git diff --exit-code -- src/docs/docs.go src/docs/swagger.json src/docs/swagger.yaml

.github/workflows/release-dev-version.yml

@@ -7,12 +7,16 @@ on:
         description: 'Version'
         required: true
 
-permissions: {}
-
+permissions:
+  contents: read
 
 jobs:
+  check-docs:
+    uses: ./.github/workflows/check-docs.yml
+
   setup:
     runs-on: ubuntu-24.04
+    needs: check-docs
     steps:
       - name: Set up QEMU
         uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0

.github/workflows/release-productive-version.yml

@@ -7,12 +7,16 @@ on:
         description: 'Version'
         required: true
 
-permissions: {}
-
+permissions:
+  contents: read
 
 jobs:
+  check-docs:
+    uses: ./.github/workflows/check-docs.yml
+
   setup:
     runs-on: ubuntu-24.04
+    needs: check-docs
     steps:
       - name: Set up QEMU
         uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0

Dockerfile

@@ -92,10 +92,21 @@ COPY src/main.go /tmp/signal-cli-rest-api-src/
 COPY src/go.mod /tmp/signal-cli-rest-api-src/
 COPY src/go.sum /tmp/signal-cli-rest-api-src/
 COPY src/plugin_loader.go /tmp/signal-cli-rest-api-src/
+COPY src/docs/add_v1_receive_schemas.go /tmp/signal-cli-rest-api-src/docs/add_v1_receive_schemas.go
+
+RUN ls -la /tmp/signal-cli-rest-api-src
+
+# build the docs
+RUN cd /tmp/signal-cli-rest-api-src && ${GOPATH}/bin/swag init --requiredByDefault --outputTypes "go,json"
+
+# manually add the json schemas for the receive V1 endpoint to the docs
+RUN cd /tmp/signal-cli-rest-api-src/docs \
+	&& wget https://github.com/Gara-Dorta/signal-cli/releases/download/v${SIGNAL_CLI_VERSION}/signal-cli-${SIGNAL_CLI_VERSION}-json-schemas.tar.gz \
+	&& mkdir signal-cli-schemas \
+	&& tar xf signal-cli-${SIGNAL_CLI_VERSION}-json-schemas.tar.gz -C signal-cli-schemas \
+	&& go run add_v1_receive_schemas.go signal-cli-schemas
 
 # build signal-cli-rest-api
-RUN ls -la /tmp/signal-cli-rest-api-src
-RUN cd /tmp/signal-cli-rest-api-src && ${GOPATH}/bin/swag init --requiredByDefault
 RUN cd /tmp/signal-cli-rest-api-src && go build -o signal-cli-rest-api main.go
 RUN cd /tmp/signal-cli-rest-api-src && go test ./client -v && go test ./utils -v
 

README.md

@@ -160,3 +160,4 @@ There are a bunch of environmental variables that can be set inside the docker c
 * `JSON_RPC_IGNORE_AVATARS`: When set to `true`, avatars are not automatically downloaded in json-rpc mode (default: `false`)
 * `JSON_RPC_IGNORE_STICKERS`: When set to `true`, sticker packs are not automatically downloaded in json-rpc mode (default: `false`)
 * `JSON_RPC_TRUST_NEW_IDENTITIES`: Choose how to trust new identities in json-rpc mode. Supported values: `on-first-use`, `always`, `never`. (default: `on-first-use`)
+* `JSON_RPC_RECEIVE_MODE`: Controls when signal-cli pulls inbound messages from the Signal servers in json-rpc mode. Supported values: `on-start` (default), `manual`. In the default `on-start` mode, signal-cli auto-receives messages and pushes them as JSON-RPC notifications regardless of whether any websocket subscriber is currently attached to `/v1/receive/{number}` — messages arriving while no subscriber is attached (e.g. during a brief subscriber redeploy) are silently dropped. In `manual` mode signal-cli only fetches messages while at least one websocket subscriber is attached; while no subscriber is attached, messages remain queued server-side under Signal's normal retention rules and are delivered on the next subscriber re-attach. Recommended for any deployment where the websocket consumer is restarted or scaled.

src/api/api.go

@@ -578,7 +578,7 @@ func (a *Api) SendV2(c *gin.Context) {
 }
 
 func (a *Api) handleSignalReceive(ws *websocket.Conn, number string, stop chan struct{}) {
-	receiveChannel, channelUuid, err := a.signalClient.GetReceiveChannel()
+	receiveChannel, channelUuid, err := a.signalClient.GetReceiveChannel(number)
 	if err != nil {
 		log.Error("Couldn't get receive channel: ", err.Error())
 		return

src/client/client.go

@@ -1067,12 +1067,12 @@ func (s *SignalClient) Receive(number string, timeout int64, ignoreAttachments b
 	}
 }
 
-func (s *SignalClient) GetReceiveChannel() (chan JsonRpc2ReceivedMessage, string, error) {
+func (s *SignalClient) GetReceiveChannel(number string) (chan JsonRpc2ReceivedMessage, string, error) {
 	jsonRpc2Client, err := s.getJsonRpc2Client()
 	if err != nil {
 		return nil, "", err
 	}
-	return jsonRpc2Client.GetReceiveChannel()
+	return jsonRpc2Client.GetReceiveChannel(number)
 }
 
 func (s *SignalClient) RemoveReceiveChannel(channelUuid string) {

src/client/jsonrpc2.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net"
 	"net/http"
 	"strconv"
@@ -56,15 +57,26 @@ func (r *RateLimitErrorType) Error() string {
 	return r.Err.Error()
 }
 
+// receiveSubscription tracks the state of a manual-mode signal-cli
+// subscribeReceive call: the subscription id assigned by signal-cli and
+// a refcount of websocket subscribers attached to that account.
+type receiveSubscription struct {
+	id       int64
+	refcount int
+}
+
 type JsonRpc2Client struct {
-	conn                     net.Conn
-	receivedResponsesById    map[string]chan JsonRpc2MessageResponse
-	receivedMessagesChannels map[string]chan JsonRpc2ReceivedMessage
-	signalCliApiConfig       *utils.SignalCliApiConfig
-	number                   string
-	receivedMessagesMutex    sync.Mutex
-	receivedResponsesMutex   sync.Mutex
-	address                  string
+	conn                       net.Conn
+	receivedResponsesById      map[string]chan JsonRpc2MessageResponse
+	receivedMessagesChannels   map[string]chan JsonRpc2ReceivedMessage
+	receiveSubscriptions       map[string]*receiveSubscription // account -> sub state
+	channelAccountByUuid       map[string]string               // channelUuid -> account
+	signalCliApiConfig         *utils.SignalCliApiConfig
+	number                     string
+	receivedMessagesMutex      sync.Mutex
+	receivedResponsesMutex     sync.Mutex
+	receiveSubscriptionsMutex  sync.Mutex
+	address                    string
 }
 
 func NewJsonRpc2Client(signalCliApiConfig *utils.SignalCliApiConfig, number string) *JsonRpc2Client {
@@ -73,6 +85,8 @@ func NewJsonRpc2Client(signalCliApiConfig *utils.SignalCliApiConfig, number stri
 		number:                   number,
 		receivedResponsesById:    make(map[string]chan JsonRpc2MessageResponse),
 		receivedMessagesChannels: make(map[string]chan JsonRpc2ReceivedMessage),
+		receiveSubscriptions:     make(map[string]*receiveSubscription),
+		channelAccountByUuid:     make(map[string]string),
 	}
 }
 
@@ -236,6 +250,19 @@ func (r *JsonRpc2Client) ReceiveData(number string, receiveWebhookUrl string) {
 		var resp1 JsonRpc2ReceivedMessage
 		json.Unmarshal([]byte(str), &resp1)
 		if resp1.Method == "receive" {
+			// In manual receive-mode signal-cli wraps the envelope in
+			// {"subscription":N,"result":{...}}; in auto mode it sends
+			// the envelope directly. Unwrap so the broadcast format is
+			// the same in both modes and downstream consumers (e.g. the
+			// websocket handler) don't have to know which mode is in use.
+			var manualWrapper struct {
+				Subscription int64           `json:"subscription"`
+				Result       json.RawMessage `json:"result"`
+			}
+			if err := json.Unmarshal(resp1.Params, &manualWrapper); err == nil && len(manualWrapper.Result) > 0 {
+				resp1.Params = manualWrapper.Result
+			}
+
 			r.receivedMessagesMutex.Lock()
 			for _, c := range r.receivedMessagesChannels {
 				select {
@@ -244,7 +271,6 @@ func (r *JsonRpc2Client) ReceiveData(number string, receiveWebhookUrl string) {
 				default:
 					log.Debug("Couldn't send message to golang channel, as there's no receiver")
 				}
-				continue
 			}
 			r.receivedMessagesMutex.Unlock()
 
@@ -270,16 +296,102 @@ func (r *JsonRpc2Client) ReceiveData(number string, receiveWebhookUrl string) {
 	}
 }
 
-func (r *JsonRpc2Client) GetReceiveChannel() (chan JsonRpc2ReceivedMessage, string, error) {
-	c := make(chan JsonRpc2ReceivedMessage)
+// subscribeReceive starts receiving messages for an account by calling
+// signal-cli's subscribeReceive JSON-RPC method. Only relevant when
+// signal-cli was launched with --receive-mode=manual; in auto mode the
+// daemon pushes notifications without an explicit subscribe call.
+// Returns the subscription id assigned by signal-cli.
+func (r *JsonRpc2Client) subscribeReceive(account string) (int64, error) {
+	type subscribeReceiveArgs struct{}
+	resultStr, err := r.getRaw("subscribeReceive", &account, subscribeReceiveArgs{})
+	if err != nil {
+		return 0, err
+	}
+	var subscriptionId int64
+	if err := json.Unmarshal([]byte(resultStr), &subscriptionId); err != nil {
+		return 0, fmt.Errorf("subscribeReceive: couldn't parse subscription id from %q: %w", resultStr, err)
+	}
+	return subscriptionId, nil
+}
+
+// unsubscribeReceive cancels a manual-mode subscription previously
+// returned by subscribeReceive.
+func (r *JsonRpc2Client) unsubscribeReceive(account string, subscriptionId int64) error {
+	type unsubscribeReceiveArgs struct {
+		Subscription int64 `json:"subscription"`
+	}
+	_, err := r.getRaw("unsubscribeReceive", &account, unsubscribeReceiveArgs{Subscription: subscriptionId})
+	return err
+}
+
+// acquireReceiveSubscription ensures an active manual-mode subscription
+// exists for the given account, refcounting concurrent websocket
+// subscribers. The first caller for an account triggers a real
+// subscribeReceive RPC; subsequent callers just bump the refcount.
+func (r *JsonRpc2Client) acquireReceiveSubscription(account string) error {
+	r.receiveSubscriptionsMutex.Lock()
+	defer r.receiveSubscriptionsMutex.Unlock()
+
+	if sub, ok := r.receiveSubscriptions[account]; ok {
+		sub.refcount++
+		return nil
+	}
+	id, err := r.subscribeReceive(account)
+	if err != nil {
+		return err
+	}
+	r.receiveSubscriptions[account] = &receiveSubscription{id: id, refcount: 1}
+	log.Infof("Subscribed to receive notifications for account %s (subscription=%d)", account, id)
+	return nil
+}
+
+// releaseReceiveSubscription decrements the per-account refcount and
+// cancels the subscription with signal-cli if it drops to zero.
+func (r *JsonRpc2Client) releaseReceiveSubscription(account string) {
+	r.receiveSubscriptionsMutex.Lock()
+	defer r.receiveSubscriptionsMutex.Unlock()
+
+	sub, ok := r.receiveSubscriptions[account]
+	if !ok {
+		return
+	}
+	sub.refcount--
+	if sub.refcount > 0 {
+		return
+	}
+	if err := r.unsubscribeReceive(account, sub.id); err != nil {
+		log.Warnf("unsubscribeReceive failed for account %s (subscription=%d): %s", account, sub.id, err.Error())
+	} else {
+		log.Infof("Unsubscribed from receive notifications for account %s (subscription=%d)", account, sub.id)
+	}
+	delete(r.receiveSubscriptions, account)
+}
+
+// GetReceiveChannel returns a channel that will receive messages for the
+// given account. If signal-cli is in manual receive-mode, it also acquires
+// a subscription so that signal-cli starts forwarding messages for the
+// account; in auto mode, account is unused (notifications flow regardless).
+//
+// account may be empty when the caller does not need a subscription
+// (e.g. legacy callers in auto mode); in that case no subscribeReceive
+// RPC is issued.
+func (r *JsonRpc2Client) GetReceiveChannel(account string) (chan JsonRpc2ReceivedMessage, string, error) {
+	c := make(chan JsonRpc2ReceivedMessage, 64)
 
 	channelUuid, err := uuid.NewV4()
 	if err != nil {
 		return c, "", err
 	}
 
+	if account != "" {
+		if err := r.acquireReceiveSubscription(account); err != nil {
+			return c, "", fmt.Errorf("subscribeReceive failed for account %s: %w", account, err)
+		}
+	}
+
 	r.receivedMessagesMutex.Lock()
 	r.receivedMessagesChannels[channelUuid.String()] = c
+	r.channelAccountByUuid[channelUuid.String()] = account
 	r.receivedMessagesMutex.Unlock()
 
 	return c, channelUuid.String(), nil
@@ -288,5 +400,11 @@ func (r *JsonRpc2Client) GetReceiveChannel() (chan JsonRpc2ReceivedMessage, stri
 func (r *JsonRpc2Client) RemoveReceiveChannel(channelUuid string) {
 	r.receivedMessagesMutex.Lock()
 	delete(r.receivedMessagesChannels, channelUuid)
+	account := r.channelAccountByUuid[channelUuid]
+	delete(r.channelAccountByUuid, channelUuid)
 	r.receivedMessagesMutex.Unlock()
+
+	if account != "" {
+		r.releaseReceiveSubscription(account)
+	}
 }

src/docs/README.md

@@ -4,44 +4,64 @@ These files are generated using the [swaggo/swag](https://github.com/swaggo/swag
 
 There are two steps, first generating the docs and then running the web server.
 
-## Generating the docs
+## With docker compose (recommended)
 
-Regenerate the files with your local source code changes.
+1. Build the docs 
+    ```bash
+    docker compose build
+    ```
+2. Serve the docs
+    ```bash
+    docker compose up
+    ```
+3. Go to http://127.0.0.1:8080/swagger/index.html to view the docs
+    * If you get a Network error, replace the IP for the docker internal IP in the error, e.g: http://172.18.0.2:8080/swagger/index.html
+
+## Locally
+
+Install [go](https://go.dev/).
+
+### Generating the docs
 
 1. Set the current working dir to `src`
     ```bash
     cd src
     ```
 1. Run swag to generate the docs
-    * Option 1, via docker 
+    * Option 1, via go
         ```bash
-        docker run --rm -v $(pwd):/code ghcr.io/swaggo/swag:latest init --requiredByDefault
+        go run github.com/swaggo/swag/cmd/swag@v1.16.6 init --requiredByDefault --outputTypes "go,json"
         ```
-    * Option 2, install swag and run the command line tool
+    * Option 2, directly with swag
         ```bash
-        swag init --requiredByDefault
+        swag init --requiredByDefault --outputTypes "go,json"
+        ```
+    * Option 3, swag via docker 
+        ```bash
+        docker run --rm -v $(pwd):/code ghcr.io/swaggo/swag:latest init --requiredByDefault --outputTypes "go,json"
+        ```
+1. Set the current working dir to `src/docs`
+    ```bash
+    cd docs
+    ```
+1. Add the signal-cli receive V1 schemas
+    * Download the `signal-cli-x.y.z-json-schemas.tar.gz` schema files from https://github.com/Gara-Dorta/signal-cli/releases
+    * Extract the files
+    * Run the script to add the schemas
+        ```bash
+        go run add_v1_receive_schemas.go ./path-to-signal-cli-json-schema-folder
         ```
 
-## Run the web server
+### Run the web server
 
 Run the web server to visualize the generated docs.
 
+1. Navigate to the `src` folder
+    ```bash
+    cd src
+    ```
 1. Run the main script
-    * Option 1, via docker, run the command at the root of the repository
-        ```bash
-        docker compose up
-        ```
-    * Option 2, install go and run the command line tool
-        ```bash
-        cd src
-        ```
-        ```bash
-        go run main.go
-        ```
-
-
-## Navigate to the docs
-
-The docs are served at: http://127.0.0.1:8080/swagger/index.html
-
-When serving with docker, if you get a Network error, replace the IP for the docker internal IP in the error, e.g: http://172.18.0.2:8080/swagger/index.html
+    ```bash
+    go run main.go
+    ```
+1. Go to http://127.0.0.1:8080/swagger/index.html

src/docs/add_v1_receive_schemas.go

@@ -0,0 +1,327 @@
+//go:build ignore
+
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	_ "github.com/bbernhard/signal-cli-rest-api/docs"
+)
+
+const (
+	goDocsPath              = "docs.go"
+	jsonDocsPath            = "swagger.json"
+	openMarker              = "const docTemplate = `"
+	closeMarker             = "`\n\n// SwaggerInfo"
+	schemesTemplateValue    = "{{ marshal .Schemes }}"
+	schemesPlaceholderToken = "__SWAG_SCHEMES_PLACEHOLDER__"
+	receivePrefix           = "receive."
+	receivePathKey          = "/v1/receive/{number}"
+	receiveWrapper          = "data.Message"
+)
+
+func main() {
+	if len(os.Args) != 2 {
+		fmt.Fprintf(os.Stderr, "usage: go run update_receive_docs.go <receiveDir>\n")
+		os.Exit(1)
+	}
+	receiveDir := os.Args[1]
+
+	if err := run(receiveDir); err != nil {
+		fmt.Fprintf(os.Stderr, "error: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run(receiveDir string) error {
+	definitions := make(map[string]interface{})
+
+	titleByFile, err := addReceiveSchemas(definitions, receiveDir)
+	if err != nil {
+		return err
+	}
+
+	updateReceiveSchemaRefs(definitions, titleByFile)
+
+	addEnvelopeWrapperDefinition(definitions)
+
+	if err := updateDocsGo(definitions); err != nil {
+		return err
+	}
+
+	if err := updateSwaggerJSON(definitions); err != nil {
+		return err
+	}
+
+	fmt.Printf("updated %s\n", goDocsPath)
+	fmt.Printf("updated %s\n", jsonDocsPath)
+	return nil
+}
+
+func updateDocsGo(receiveDefinitions map[string]interface{}) error {
+	content, err := os.ReadFile(goDocsPath)
+	if err != nil {
+		return fmt.Errorf("read %s: %w", goDocsPath, err)
+	}
+
+	template, templateStart, templateEnd, err := extractDocTemplate(string(content))
+	if err != nil {
+		return err
+	}
+
+	updatedTemplate, err := updateJSONDocument(toValidJson(template), receiveDefinitions)
+	if err != nil {
+		return err
+	}
+	updatedTemplate = encodeForGoRawString(updatedTemplate)
+
+	updated := string(content[:templateStart]) + updatedTemplate + string(content[templateEnd:])
+	if err := os.WriteFile(goDocsPath, []byte(updated), 0644); err != nil {
+		return fmt.Errorf("write %s: %w", goDocsPath, err)
+	}
+
+	return nil
+}
+
+func updateSwaggerJSON(receiveDefinitions map[string]interface{}) error {
+	content, err := os.ReadFile(jsonDocsPath)
+	if err != nil {
+		return fmt.Errorf("read %s: %w", jsonDocsPath, err)
+	}
+
+	updated, err := updateJSONDocument(string(content), receiveDefinitions)
+	if err != nil {
+		return err
+	}
+
+	if err := os.WriteFile(jsonDocsPath, []byte(updated), 0644); err != nil {
+		return fmt.Errorf("write %s: %w", jsonDocsPath, err)
+	}
+
+	return nil
+}
+
+func extractDocTemplate(content string) (string, int, int, error) {
+	start := strings.Index(content, openMarker)
+	if start == -1 {
+		return "", -1, -1, fmt.Errorf("could not find docTemplate start in %s", goDocsPath)
+	}
+
+	start += len(openMarker)
+	endOffset := strings.Index(content[start:], closeMarker)
+	if endOffset == -1 {
+		return "", -1, -1, fmt.Errorf("could not find docTemplate end in %s", goDocsPath)
+	}
+
+	end := start + endOffset
+	return content[start:end], start, end, nil
+}
+
+func toValidJson(content string) string {
+	content = strings.ReplaceAll(content, "` + \"`\" + `", "`")
+	content = strings.Replace(content, schemesTemplateValue, `"`+schemesPlaceholderToken+`"`, 1)
+	return content
+}
+
+func encodeForGoRawString(content string) string {
+	content = strings.ReplaceAll(content, "`", "` + \"`\" + `")
+	content = strings.Replace(content, `"`+schemesPlaceholderToken+`"`, schemesTemplateValue, 1)
+	return content
+}
+
+func updateJSONDocument(content string, receiveDefinitions map[string]interface{}) (string, error) {
+	var document map[string]interface{}
+	if err := json.Unmarshal([]byte(content), &document); err != nil {
+		return "", fmt.Errorf("parse document: %w", err)
+	}
+
+	if err := applyReceiveSchemaUpdates(document, receiveDefinitions); err != nil {
+		return "", err
+	}
+
+	raw, err := json.MarshalIndent(document, "", "    ")
+	if err != nil {
+		return "", fmt.Errorf("marshal document: %w", err)
+	}
+
+	return string(raw), nil
+}
+
+func addReceiveSchemas(definitions map[string]interface{}, receiveDir string) (map[string]string, error) {
+	entries, err := os.ReadDir(receiveDir)
+	if err != nil {
+		return nil, fmt.Errorf("read %s: %w", receiveDir, err)
+	}
+
+	files := make([]string, 0)
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".schema.json") {
+			continue
+		}
+		files = append(files, entry.Name())
+	}
+	sort.Strings(files)
+	titleByFile := make(map[string]string, len(files))
+
+	for _, name := range files {
+		fullPath := filepath.Join(receiveDir, name)
+		data, err := os.ReadFile(fullPath)
+		if err != nil {
+			return nil, fmt.Errorf("read schema file %s: %w", fullPath, err)
+		}
+
+		var schemaObj map[string]interface{}
+		if err := json.Unmarshal(data, &schemaObj); err != nil {
+			return nil, fmt.Errorf("parse schema file %s: %w", fullPath, err)
+		}
+
+		title, ok := schemaObj["title"].(string)
+		if !ok || strings.TrimSpace(title) == "" {
+			return nil, fmt.Errorf("schema file %s is missing title", fullPath)
+		}
+
+		titleByFile[name] = title
+		definitions[receivePrefix+title] = removeSchemaKeysRecursive(schemaObj, "")
+	}
+
+	return titleByFile, nil
+}
+
+func removeSchemaKeysRecursive(value interface{}, parentKey string) interface{} {
+	switch typed := value.(type) {
+	case map[string]interface{}:
+		updated := make(map[string]interface{}, len(typed))
+		for key, item := range typed {
+			if key == "$schema" || key == "$id" {
+				continue
+			}
+			if key == "title" && parentKey != "properties" {
+				continue
+			}
+			updated[key] = removeSchemaKeysRecursive(item, key)
+		}
+		return updated
+	case []interface{}:
+		updated := make([]interface{}, len(typed))
+		for idx, item := range typed {
+			updated[idx] = removeSchemaKeysRecursive(item, parentKey)
+		}
+		return updated
+	default:
+		return value
+	}
+}
+
+func updateReceiveSchemaRefs(definitions map[string]interface{}, titleByFile map[string]string) {
+	for key, value := range definitions {
+		if !strings.HasPrefix(key, receivePrefix) {
+			continue
+		}
+		definitions[key] = rewriteSchemaRefs(value, titleByFile)
+	}
+}
+
+func rewriteSchemaRefs(value interface{}, titleByFile map[string]string) interface{} {
+	switch typed := value.(type) {
+	case map[string]interface{}:
+		updated := make(map[string]interface{}, len(typed))
+		for k, v := range typed {
+			if k == "$ref" {
+				if refValue, ok := v.(string); ok {
+					if strings.HasSuffix(refValue, ".schema.json") {
+						base := filepath.Base(refValue)
+						if title, exists := titleByFile[base]; exists {
+							updated[k] = "#/definitions/" + receivePrefix + title
+							continue
+						}
+					}
+				}
+			}
+
+			updated[k] = rewriteSchemaRefs(v, titleByFile)
+		}
+		return updated
+	case []interface{}:
+		updated := make([]interface{}, len(typed))
+		for idx, item := range typed {
+			updated[idx] = rewriteSchemaRefs(item, titleByFile)
+		}
+		return updated
+	default:
+		return value
+	}
+}
+
+func addEnvelopeWrapperDefinition(definitions map[string]interface{}) {
+	definitions[receiveWrapper] = map[string]interface{}{
+		"type": "object",
+		"properties": map[string]interface{}{
+			"account":  map[string]interface{}{"type": "string"},
+			"envelope": map[string]interface{}{"$ref": "#/definitions/receive.MessageEnvelope"},
+		},
+		"required": []interface{}{"account", "envelope"},
+	}
+}
+
+func applyReceiveSchemaUpdates(document map[string]interface{}, receiveDefinitions map[string]interface{}) error {
+	definitions, err := getObject(document, "definitions")
+	if err != nil {
+		return err
+	}
+
+	for key := range definitions {
+		if strings.HasPrefix(key, receivePrefix) || key == receiveWrapper {
+			delete(definitions, key)
+		}
+	}
+
+	for key, value := range receiveDefinitions {
+		definitions[key] = value
+	}
+
+	paths, err := getObject(document, "paths")
+	if err != nil {
+		return err
+	}
+	receivePath, err := getObject(paths, receivePathKey)
+	if err != nil {
+		return err
+	}
+	receiveGet, err := getObject(receivePath, "get")
+	if err != nil {
+		return err
+	}
+	responses, err := getObject(receiveGet, "responses")
+	if err != nil {
+		return err
+	}
+	response200, err := getObject(responses, "200")
+	if err != nil {
+		return err
+	}
+
+	response200["schema"] = map[string]interface{}{
+		"$ref": "#/definitions/" + receiveWrapper,
+	}
+
+	return nil
+}
+
+func getObject(parent map[string]interface{}, key string) (map[string]interface{}, error) {
+	value, ok := parent[key]
+	if !ok {
+		return nil, fmt.Errorf("missing key %q", key)
+	}
+
+	obj, ok := value.(map[string]interface{})
+	if !ok {
+		return nil, fmt.Errorf("key %q is not an object", key)
+	}
+
+	return obj, nil
+}

src/scripts/jsonrpc2-helper.go

@@ -13,7 +13,7 @@ import (
 const supervisorctlConfigTemplate = `
 [program:%s]
 process_name=%s
-command=%s --output=json --config %s%s daemon %s%s%s%s --tcp 127.0.0.1:%d
+command=%s --output=json --config %s%s daemon%s %s%s%s%s --tcp 127.0.0.1:%d
 autostart=true
 autorestart=true
 startretries=10
@@ -75,6 +75,22 @@ func main() {
 		signalCliIgnoreStickers = " --ignore-stickers"
 	}
 
+	// Receive mode: by default signal-cli auto-receives messages on the
+	// daemon and pushes them as JSON-RPC notifications, regardless of
+	// whether any websocket subscriber is connected. In a deploy where
+	// the subscribing client briefly disconnects, those notifications
+	// have nowhere to go and are silently dropped. With manual receive
+	// mode, signal-cli only fetches messages when explicitly requested
+	// via subscribeReceive — see jsonrpc2.go — so messages stay buffered
+	// on the Signal servers until a subscriber re-attaches.
+	signalCliReceiveMode := ""
+	receiveMode := utils.GetEnv("JSON_RPC_RECEIVE_MODE", "")
+	if receiveMode == "manual" {
+		signalCliReceiveMode = " --receive-mode=manual"
+	} else if receiveMode != "" && receiveMode != "on-start" {
+		log.Fatal("Invalid JSON_RPC_RECEIVE_MODE environment variable set! Must be 'manual' or 'on-start'.")
+	}
+
 	supervisorctlProgramName := "signal-cli-json-rpc-1"
 	supervisorctlLogFolder := "/var/log/" + supervisorctlProgramName
 	_, err := exec.Command("mkdir", "-p", supervisorctlLogFolder).Output()
@@ -100,7 +116,7 @@ func main() {
 	supervisorctlConfigFilename := "/etc/supervisor/conf.d/" + "signal-cli-json-rpc-1.conf"
 
 	supervisorctlConfig := fmt.Sprintf(supervisorctlConfigTemplate, supervisorctlProgramName, supervisorctlProgramName, signalCliBinary,
-		signalCliConfigDir, trustNewIdentities, signalCliIgnoreAttachments, signalCliIgnoreStories,
+		signalCliConfigDir, trustNewIdentities, signalCliReceiveMode, signalCliIgnoreAttachments, signalCliIgnoreStories,
 		signalCliIgnoreAvatars, signalCliIgnoreStickers, tcpPort,
 		supervisorctlProgramName, supervisorctlProgramName)