Finalize immutable message hashes before delivery

.github/workflows/go1.25.yml

@@ -10,6 +10,20 @@ jobs:
 
   build:
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17-alpine
+        env:
+          POSTGRES_HOST_AUTH_METHOD: trust
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      FMSG_TEST_DATABASE_URL: postgres://postgres@localhost:5432/postgres?sslmode=disable
     steps:
     - uses: actions/checkout@v3
 
@@ -22,4 +36,4 @@ jobs:
       run: go build -v ./...
 
     - name: Test
-      run: go test -v ./...
+      run: go test -race ./...

.github/workflows/integration-test.yml

@@ -11,15 +11,26 @@ jobs:
       - name: Trigger and wait for integration test
         env:
           GH_TOKEN: ${{ secrets.FMSG_DOCKER_PAT }}
+          PR_BRANCH: ${{ github.event.pull_request.head.ref }}
         run: |
-          FMSGD_REF="${{ github.event.pull_request.head.ref }}"
+          FMSGD_REF="$PR_BRANCH"
+          matching_ref() {
+            if gh api "repos/markmnl/$1/git/ref/heads/$PR_BRANCH" >/dev/null 2>&1; then
+              printf '%s\n' "$PR_BRANCH"
+            else
+              echo main
+            fi
+          }
+          DOCKER_REF=$(matching_ref fmsg-docker)
+          WEBAPI_REF=$(matching_ref fmsg-webapi)
+          DISPATCH_STARTED=$(date -u +%Y-%m-%dT%H:%M:%SZ)
 
-          # Trigger the integration test workflow in fmsg-docker,
-          # passing the PR branch so it builds fmsgd from the PR
+          # Coordinated schema/API changes use their companion branches.
           gh workflow run integration-test.yml \
             --repo markmnl/fmsg-docker \
-            --ref main \
-            -f fmsgd_ref="$FMSGD_REF"
+            --ref "$DOCKER_REF" \
+            -f fmsgd_ref="$FMSGD_REF" \
+            -f fmsg_webapi_ref="$WEBAPI_REF"
 
           echo "Triggered integration test for fmsgd_ref=$FMSGD_REF, polling for run..."
 
@@ -30,9 +41,12 @@ jobs:
             RUN_ID=$(gh run list \
               --repo markmnl/fmsg-docker \
               --workflow integration-test.yml \
+              --event workflow_dispatch \
+              --branch "$DOCKER_REF" \
+              --created ">=$DISPATCH_STARTED" \
               --limit 10 \
               --json databaseId,displayTitle \
-              --jq ".[] | select(.displayTitle | contains(\"$FMSGD_REF\")) | .databaseId" \
+            | jq -r --arg ref "$FMSGD_REF" '.[] | select(.displayTitle | contains($ref)) | .databaseId' \
             | head -1)
             if [ -n "$RUN_ID" ]; then
               break

README.md

@@ -139,4 +139,64 @@ PGDATABASE=fmsgd
 sudo systemctl daemon-reload
 sudo systemctl enable fmsgd
 sudo systemctl start fmsgd
-```
+```
+## Immutable message finalization and upgrades
+
+`fmsg-webapi` finalizes local messages with `pkg/message`: the timestamp, SHA-256,
+exact header, and durable wire payloads are committed together, including local-only
+messages and reactions. The hash covers the encoded wire header and expanded body
+and attachment bytes. Compression and common media type encoding are chosen before
+hashing. Add-to exchanges retain independent hashes and reuse the finalized payload.
+The daemon reuses these representations for federation and challenge responses;
+it refuses a representation that differs from an established hash.
+
+The `wire_message` JSONB columns are versioned internal snapshots containing payload
+paths; they are not API objects. `.fmsg-wire-*` directories beside message content
+must be retained with the message database and data directory. Both services need
+access to the shared files (normally the same service user/group). The API keeps its
+expanded downloadable content separately. New received messages also preserve their
+wire payloads before expanding the downloadable copies.
+
+`dd.sql` bootstraps a new, empty database. The daemon and API require finalized
+sent messages and do not repair old rows during normal operation.
+
+For an existing installation, build the single standalone migration binary:
+
+```sh
+CGO_ENABLED=0 go build -o fmsg-backfill ./cmd/fmsg-backfill
+```
+
+The binary embeds the schema changes; no SQL scripts, source checkout or running
+services are needed on the target host. It upgrades the pre-finalization schema
+(with `wire_header` and add-to batch hashes) and can also verify a completed migration.
+It uses the standard `PG*` connection variables. Run it as the service account with
+write access to the database and every stored payload path, including shared volumes.
+
+Stop both services and back up the message database and data directory together.
+Then validate and apply the conversion before starting the matching daemon and API:
+
+```sh
+./fmsg-backfill -domain example.com
+./fmsg-backfill -domain example.com -apply
+```
+
+The default is a full dry run: it reconstructs and verifies every sent message and
+batch, then rolls back schema/data changes and removes staged files. `-apply` commits
+the schema and data together in one transaction. It preserves timestamps, message IDs
+and all published hashes, finalizes local-only parents before replies, and prepares
+existing federated messages for later delivery. A successful run installs the strict
+schema; **do not rerun `dd.sql` against the existing database**.
+
+Missing files, inconsistent reply identities or representations that cannot reproduce
+a published hash fail the entire migration. Old received compression must be
+reconstructible with the exact declared wire size; otherwise recover the original wire
+payload before upgrading. Resolve reported records and rerun while services remain
+stopped. The command is separate from the daemon and is not bundled in its image.
+A process crash before commit may leave an unreferenced `.fmsg-wire-*` directory;
+only remove such directories after checking both snapshot columns for references.
+
+PostgreSQL tests use an isolated temporary schema in the supplied test database:
+
+```sh
+FMSG_TEST_DATABASE_URL=postgres://postgres@localhost/fmsg_test?sslmode=disable go test ./...
+```

cmd/fmsg-backfill/main.go

@@ -0,0 +1,118 @@
+// fmsg-backfill upgrades the pre-finalization message store offline. All legacy
+// reconstruction and schema conversion live in this standalone command.
+package main
+
+import (
+	"context"
+	"database/sql"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"regexp"
+	"strings"
+
+	_ "github.com/lib/pq"
+	"github.com/markmnl/fmsgd"
+)
+
+func main() {
+	domain := flag.String("domain", "", "local sending domain (required)")
+	apply := flag.Bool("apply", false, "commit schema and data conversion; default validates then rolls back")
+	flag.Parse()
+	if *domain == "" {
+		log.Fatal("-domain is required")
+	}
+	db, err := sql.Open("postgres", "") // standard PG* environment variables
+	if err == nil {
+		defer db.Close()
+		err = migrate(context.Background(), db, *domain, *apply, os.Stdout)
+	}
+	if err != nil {
+		log.Print(err)
+		os.Exit(1)
+	}
+}
+
+// One transaction owns both the schema change and every converted row. The
+// operator stops services first; NOWAIT also refuses a store still in use.
+func migrate(ctx context.Context, db *sql.DB, domain string, apply bool, out io.Writer) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	m := &migration{tx: tx, domain: domain, visiting: make(map[int64]bool), done: make(map[int64]bool)}
+	commitAttempted := false
+	defer func() {
+		_ = tx.Rollback()
+		// A lost COMMIT acknowledgement does not prove rollback. Retain the
+		// files in that case and let the next run verify committed snapshots.
+		if !commitAttempted {
+			for _, dir := range m.files {
+				_ = os.RemoveAll(dir)
+			}
+		}
+	}()
+	if _, err = tx.ExecContext(ctx, `LOCK TABLE msg,msg_to,msg_attachment,msg_add_to_batch,msg_add_to,msg_add_to_notify IN ACCESS EXCLUSIVE MODE NOWAIT`); err != nil {
+		return fmt.Errorf("stop daemon and API before migration: %w", err)
+	}
+	// Bootstrap SQL remains plain CREATE statements. Only this command knows
+	// the previous schema and how to replace its triggers in place.
+	_, functions, ok := strings.Cut(fmsgd.Schema, "-- Functions and triggers.\n")
+	if !ok {
+		return fmt.Errorf("embedded schema has no functions section")
+	}
+	triggerPattern := regexp.MustCompile(`(?s)create (?:constraint )?trigger (\w+)\s+.*?\bon (\w+)\s`)
+	for _, match := range triggerPattern.FindAllStringSubmatch(functions, -1) {
+		if _, err = tx.ExecContext(ctx, "DROP TRIGGER IF EXISTS "+match[1]+" ON "+match[2]); err != nil {
+			return err
+		}
+	}
+	if _, err = tx.ExecContext(ctx, `
+		DROP TRIGGER IF EXISTS trg_msg_prevent_unreferenceable_parent ON msg;
+		DROP FUNCTION IF EXISTS prevent_referenced_msg_from_becoming_unreferenceable();
+		ALTER TABLE msg ADD COLUMN IF NOT EXISTS wire_message jsonb;
+		ALTER TABLE msg_add_to_batch ADD COLUMN IF NOT EXISTS wire_message jsonb;
+		CREATE INDEX IF NOT EXISTS msg_add_to_batch_sha256_idx ON msg_add_to_batch (sha256) WHERE sha256 IS NOT NULL;
+		CREATE INDEX IF NOT EXISTS msg_pid_idx ON msg (pid) WHERE pid IS NOT NULL;
+	`); err != nil {
+		return err
+	}
+	ids, err := m.ids(`SELECT id FROM msg WHERE time_sent IS NOT NULL ORDER BY id`)
+	if err != nil {
+		return err
+	}
+	for _, id := range ids {
+		if err = m.message(id); err != nil {
+			return fmt.Errorf("message %d: %w; database changes rolled back", id, err)
+		}
+		fmt.Fprintf(out, "verified message %d\n", id)
+	}
+	// Draft children may have existed before their local parent had a hash.
+	if _, err = tx.ExecContext(ctx, `UPDATE msg child SET psha256=parent.sha256 FROM msg parent WHERE child.pid=parent.id AND child.psha256 IS NULL AND child.time_sent IS NULL`); err != nil {
+		return err
+	}
+	if err = m.validate(); err != nil {
+		return err
+	}
+	// Replace function definitions only here, without duplicating them or
+	// carrying upgrade statements in the bootstrap schema.
+	functions = strings.ReplaceAll(functions, "create function ", "create or replace function ")
+	if _, err = tx.ExecContext(ctx, functions); err != nil {
+		return err
+	}
+	if !apply {
+		if err = tx.Rollback(); err != nil {
+			return err
+		}
+		fmt.Fprintln(out, "Dry run passed; schema, data and staged files rolled back. Run with -apply to commit.")
+		return nil
+	}
+	commitAttempted = true
+	if err = tx.Commit(); err != nil {
+		return fmt.Errorf("commit outcome uncertain; keep payload files and rerun to verify: %w", err)
+	}
+	fmt.Fprintln(out, "Migration committed. Start the matching daemon and API; do not rerun dd.sql.")
+	return nil
+}