diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml
index 685e26c..49e8de3 100644
--- a/.github/workflows/build-test.yml
+++ b/.github/workflows/build-test.yml
@@ -8,6 +8,20 @@ on:
 jobs:
   build-and-test:
     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@v4
 
@@ -19,4 +33,5 @@ jobs:
         run: go build ./...
 
       - name: Test
-        run: go test ./...
+        run: |
+          FMSG_TEST_DD="$(go list -m -f '{{.Dir}}' github.com/markmnl/fmsgd)/dd.sql" go test -race ./...
diff --git a/README.md b/README.md
index d8a8bf3..24300d6 100644
--- a/README.md
+++ b/README.md
@@ -633,9 +633,33 @@ Deletes a draft message and all its attachments from the database and disk. Only
 | `403`  | Not the owner, or message already sent |
 | `404`  | Message not found |
 
+### Message hashes and references
+
+A sent or received message includes `sha256`, a lowercase 64-character hexadecimal
+protocol hash; a draft returns `null`. `psha256` is the exact parent hash or `null`.
+Both fields appear in message detail, inbox/sent lists and WebSocket message
+payloads. Structured thread entries include them too, retaining `message_sha256`
+as an alias for existing clients. Each `add_to` batch exposes its own `sha256`.
+These hashes describe the protocol message, not the API JSON or mutable read and
+delivery state.
+
+Every message `:id` route accepts either its positive numeric ID or its 64-character
+SHA-256 (hexadecimal input is case-insensitive). This includes detail, body and
+attachment downloads, thread routes, read, react, and add-to. Hashes do not grant
+access: the same identity checks apply. Invalid references return `400`; unknown
+hashes return `404`. Drafts have no hash, and draft-only operations still reject
+sent messages.
+
+Create/update accepts `pid` as a numeric parent ID or a hexadecimal hash string.
+Response `pid` remains numeric. The hash may identify an original message or an
+add-to batch; a batch reference is retained exactly in `psha256`. A participant
+added only through a batch must use that batch's hash when replying. The parent
+must be sent and non-terminal, and the caller must participate in that specific
+original or batch. Client-provided `sha256`/`psha256` values are not trusted.
+
 ### POST `/fmsg/:id/send`
 
-Marks a draft message as sent by setting `time_sent` to the current timestamp. Only the owner may send.
+Atomically stamps `time_sent`, computes the protocol `sha256`, and retains the finalized wire representation. This happens for local-only delivery too. Only the owner may send. Compression and common media type encoding are chosen before hashing; future federation reuses that representation. A failure leaves the message a draft. Concurrent edits, attachment changes, deletion, and send are serialized; sent content is immutable.
 
 **Send is the validation gate.** Drafts are a workspace: create/update accept
 incomplete messages (no recipients, no type) by design, but a draft may only
@@ -657,7 +681,7 @@ third-party delivery's outcome, so the reply is attempted and the wire's
 originating domain always passes (it retains its outgoing messages). Local
 recipients are unaffected.
 
-**Response:** `200 OK` with `{"id": <int>, "time": <float64>}`.
+**Response:** `200 OK` with `{"id": <int>, "time": <float64>, "sha256": "<64 hex characters>"}`.
 
 **Errors:**
 
@@ -697,7 +721,7 @@ This endpoint records the add-to as a new `msg_add_to_batch` row (capturing the
 
 New addresses must be distinct among themselves (case-insensitive).
 
-**Response:** `200 OK` with `{"id": <int>, "added": <int>}`.
+**Response:** `200 OK` with `{"id": <int>, "added": <int>, "batch_id": <int>, "sha256": "<64 hex characters>"}`. On a draft, the batch hash is `null` until send finalizes it.
 
 **Errors:**
 
@@ -731,7 +755,7 @@ Setting the same reaction the caller already has is idempotent: nothing is
 sent and the existing reaction message is returned with `200 OK`. Clearing when
 the caller has no reaction returns `200 OK` with `null` values.
 
-**Response:** `201 Created` with the reaction message's `{"id": <int>, "time": <number>}`.
+**Response:** `201 Created` with the reaction message's `{"id": <int>, "time": <number>, "sha256": "<64 hex characters>"}`.
 
 **Errors:**
 
@@ -815,8 +839,8 @@ truncated.
       "to": ["@agent@example.net"],
       "type": "text/plain",
       "size": 5,
-      "message_sha256": "0123456789abcdef",
-      "body": {"type": "text/plain", "size": 5, "text": "hello", "cache_key": "sha256:0123456789abcdef:body", "cacheable": true},
+      "message_sha256": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+      "body": {"type": "text/plain", "size": 5, "text": "hello", "cache_key": "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef:body", "cacheable": true},
       "attachments": []
     }
   ]
@@ -967,3 +991,23 @@ CREATE TABLE push_subscription (
     PRIMARY KEY (addr, endpoint)
 );
 ```
+
+### Finalization schema upgrade and PostgreSQL tests
+
+This version requires the matching `fmsgd/dd.sql` schema and finalization-capable
+daemon. For existing installations, stop both services and run the standalone
+`fmsg-backfill` binary before starting the new versions. The binary embeds the schema
+upgrade, prepares existing sent messages and batches, and backfills local-only hashes
+in one transaction. Run without `-apply` for a full dry run, then with `-apply` to commit.
+Do not rerun the daemon's bootstrap `dd.sql` on an existing database. See
+[the daemon upgrade instructions](https://github.com/markmnl/fmsgd#immutable-message-finalization-and-upgrades).
+All sent messages have hashes; only drafts expose a `null` message hash.
+
+`go test ./...` runs the unit tests. PostgreSQL integration tests create and remove
+isolated schemas; set `FMSG_TEST_DATABASE_URL` to a disposable test database and
+`FMSG_TEST_DD` to the matching daemon schema (defaults to the sibling checkout):
+
+```sh
+FMSG_TEST_DATABASE_URL='postgres://postgres@localhost/fmsg_test?sslmode=disable' \
+FMSG_TEST_DD=/path/to/fmsgd/dd.sql go test ./...
+```
diff --git a/cmd/fmsg-webapi/main.go b/cmd/fmsg-webapi/main.go
index 48512ca..dcc1f5c 100644
--- a/cmd/fmsg-webapi/main.go
+++ b/cmd/fmsg-webapi/main.go
@@ -185,21 +185,21 @@ func main() {
 
 		fmsg.GET("", msgHandler.List)
 		fmsg.GET("/sent", msgHandler.Sent)
-		fmsg.POST("", msgHandler.Create)
+		fmsg.POST("", msgHandler.Atomic((*handlers.MessageHandler).Create))
 		fmsg.GET("/:id", msgHandler.Get)
-		fmsg.PUT("/:id", msgHandler.Update)
-		fmsg.DELETE("/:id", msgHandler.Delete)
-		fmsg.POST("/:id/send", msgHandler.Send)
+		fmsg.PUT("/:id", msgHandler.Atomic((*handlers.MessageHandler).Update))
+		fmsg.DELETE("/:id", msgHandler.Atomic((*handlers.MessageHandler).Delete))
+		fmsg.POST("/:id/send", msgHandler.Atomic((*handlers.MessageHandler).Send))
 		fmsg.POST("/:id/read", msgHandler.MarkRead)
-		fmsg.POST("/:id/add-to", msgHandler.AddRecipients)
-		fmsg.POST("/:id/react", msgHandler.React)
+		fmsg.POST("/:id/add-to", msgHandler.Atomic((*handlers.MessageHandler).AddRecipients))
+		fmsg.POST("/:id/react", msgHandler.Atomic((*handlers.MessageHandler).React))
 		fmsg.GET("/:id/data", msgHandler.DownloadData)
 		fmsg.GET("/:id/thread", msgHandler.ThreadText)
 		fmsg.GET("/:id/thread/messages", msgHandler.ThreadMessages)
 
-		fmsg.POST("/:id/attach", attHandler.Upload)
+		fmsg.POST("/:id/attach", attHandler.Atomic((*handlers.AttachmentHandler).Upload))
 		fmsg.GET("/:id/attach/:filename", attHandler.Download)
-		fmsg.DELETE("/:id/attach/:filename", attHandler.DeleteAttachment)
+		fmsg.DELETE("/:id/attach/:filename", attHandler.Atomic((*handlers.AttachmentHandler).DeleteAttachment))
 
 		if pushHandler != nil {
 			fmsg.POST("/push/subscribe", pushHandler.Subscribe)
diff --git a/go.mod b/go.mod
index 5f2d865..87d5dbd 100644
--- a/go.mod
+++ b/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/jackc/pgx/v5 v5.8.0
 	github.com/joho/godotenv v1.5.1
+	github.com/markmnl/fmsgd v0.5.1-0.20260910081105-17cb088bca99
 	golang.org/x/sync v0.20.0
 )
 
diff --git a/go.sum b/go.sum
index 14eaf0a..36a5425 100644
--- a/go.sum
+++ b/go.sum
@@ -58,6 +58,10 @@ github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzh
 github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
 github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/markmnl/fmsgd v0.5.1-0.20260910081105-17cb088bca99 h1:LTzLNjH5hKj5X9YfVrJotqn3dMH3UJ0QQYIFrLzmIaQ=
+github.com/markmnl/fmsgd v0.5.1-0.20260910081105-17cb088bca99/go.mod h1:DNykk/1Z25IM+yzSxjtJs51nEYYaA1H03dAMG37JERk=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
diff --git a/internal/handlers/attachments.go b/internal/handlers/attachments.go
index ea7a922..951645f 100644
--- a/internal/handlers/attachments.go
+++ b/internal/handlers/attachments.go
@@ -19,6 +19,7 @@ import (
 
 // AttachmentHandler holds dependencies for attachment routes.
 type AttachmentHandler struct {
+	query         queries
 	DB            *db.DB
 	DataDir       string
 	MaxAttachSize int64
@@ -33,7 +34,7 @@ func NewAttachmentHandler(database *db.DB, dataDir string, maxAttachSize, maxMsg
 // Upload handles POST /fmsg/:id/attachments.
 func (h *AttachmentHandler) Upload(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -43,7 +44,7 @@ func (h *AttachmentHandler) Upload(c *gin.Context) {
 	// Load the message to check ownership and draft status.
 	var fromAddr string
 	var timeSent *float64
-	err := h.DB.Pool.QueryRow(ctx,
+	err := h.q().QueryRow(ctx,
 		"SELECT from_addr, time_sent FROM msg WHERE id = $1", msgID,
 	).Scan(&fromAddr, &timeSent)
 	if err != nil {
@@ -113,9 +114,10 @@ func (h *AttachmentHandler) Upload(c *gin.Context) {
 		return
 	}
 
+	createdFile(c, finalPath)
 	// Check total message size and persist attachment in a transaction to
 	// prevent concurrent uploads from exceeding MaxMsgSize.
-	tx, err := h.DB.Pool.Begin(ctx)
+	tx, err := h.q().Begin(ctx)
 	if err != nil {
 		_ = os.Remove(finalPath)
 		log.Printf("upload attachment: begin tx: %v", err)
@@ -169,7 +171,7 @@ func (h *AttachmentHandler) Upload(c *gin.Context) {
 // Download handles GET /fmsg/:id/attachments/:filename.
 func (h *AttachmentHandler) Download(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -185,7 +187,7 @@ func (h *AttachmentHandler) Download(c *gin.Context) {
 
 	// Check ownership or recipient access.
 	var fromAddr string
-	err := h.DB.Pool.QueryRow(ctx, "SELECT from_addr FROM msg WHERE id = $1", msgID).Scan(&fromAddr)
+	err := h.q().QueryRow(ctx, "SELECT from_addr FROM msg WHERE id = $1", msgID).Scan(&fromAddr)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
 			c.JSON(http.StatusNotFound, gin.H{"error": "message not found"})
@@ -199,7 +201,7 @@ func (h *AttachmentHandler) Download(c *gin.Context) {
 	// Check recipients (to or add_to) if not owner.
 	if fromAddr != identity {
 		var recipientCount int
-		if err = h.DB.Pool.QueryRow(ctx,
+		if err = h.q().QueryRow(ctx,
 			`SELECT COUNT(*) FROM (
 				SELECT 1 FROM msg_to WHERE msg_id = $1 AND lower(addr) = lower($2)
 				UNION ALL
@@ -213,7 +215,7 @@ func (h *AttachmentHandler) Download(c *gin.Context) {
 
 	// Look up the attachment filepath.
 	var storedPath string
-	err = h.DB.Pool.QueryRow(ctx,
+	err = h.q().QueryRow(ctx,
 		"SELECT filepath FROM msg_attachment WHERE msg_id = $1 AND filename = $2",
 		msgID, filename,
 	).Scan(&storedPath)
@@ -242,7 +244,7 @@ func (h *AttachmentHandler) Download(c *gin.Context) {
 // DeleteAttachment handles DELETE /fmsg/:id/attachments/:filename.
 func (h *AttachmentHandler) DeleteAttachment(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -257,7 +259,7 @@ func (h *AttachmentHandler) DeleteAttachment(c *gin.Context) {
 
 	var fromAddr string
 	var timeSent *float64
-	err := h.DB.Pool.QueryRow(ctx,
+	err := h.q().QueryRow(ctx,
 		"SELECT from_addr, time_sent FROM msg WHERE id = $1", msgID,
 	).Scan(&fromAddr, &timeSent)
 	if err != nil {
@@ -281,7 +283,7 @@ func (h *AttachmentHandler) DeleteAttachment(c *gin.Context) {
 
 	// Get filepath before deleting.
 	var storedPath string
-	err = h.DB.Pool.QueryRow(ctx,
+	err = h.q().QueryRow(ctx,
 		"SELECT filepath FROM msg_attachment WHERE msg_id = $1 AND filename = $2",
 		msgID, filename,
 	).Scan(&storedPath)
@@ -295,7 +297,7 @@ func (h *AttachmentHandler) DeleteAttachment(c *gin.Context) {
 		return
 	}
 
-	if _, err = h.DB.Pool.Exec(ctx,
+	if _, err = h.q().Exec(ctx,
 		"DELETE FROM msg_attachment WHERE msg_id = $1 AND filename = $2", msgID, filename,
 	); err != nil {
 		log.Printf("delete attachment: db: %v", err)
@@ -307,7 +309,7 @@ func (h *AttachmentHandler) DeleteAttachment(c *gin.Context) {
 	cleanPath := filepath.Clean(storedPath)
 	cleanDataDir := filepath.Clean(h.DataDir)
 	if strings.HasPrefix(cleanPath, cleanDataDir+string(filepath.Separator)) {
-		_ = os.Remove(cleanPath)
+		afterCommit(c, func() { _ = os.Remove(cleanPath) })
 	}
 
 	c.Status(http.StatusNoContent)
diff --git a/internal/handlers/finalization_integration_test.go b/internal/handlers/finalization_integration_test.go
new file mode 100644
index 0000000..5305cdd
--- /dev/null
+++ b/internal/handlers/finalization_integration_test.go
@@ -0,0 +1,335 @@
+package handlers
+
+import (
+	"bytes"
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/markmnl/fmsg-webapi/internal/db"
+	"github.com/markmnl/fmsg-webapi/internal/middleware"
+	"github.com/markmnl/fmsgd/pkg/fmsg"
+)
+
+type finalizationAPI struct {
+	router *gin.Engine
+	h      *MessageHandler
+	pool   *pgxpool.Pool
+}
+
+func newFinalizationAPI(t *testing.T) *finalizationAPI {
+	t.Helper()
+	dsn := os.Getenv("FMSG_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("set FMSG_TEST_DATABASE_URL to run PostgreSQL integration tests")
+	}
+	ctx := context.Background()
+	admin, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	schema := fmt.Sprintf("hash_api_%d", time.Now().UnixNano())
+	if _, err = admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = admin.Exec(ctx, "DROP SCHEMA "+schema+" CASCADE"); admin.Close() })
+	config, err := pgxpool.ParseConfig(dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	config.ConnConfig.RuntimeParams["search_path"] = schema
+	pool, err := pgxpool.NewWithConfig(ctx, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	dd := os.Getenv("FMSG_TEST_DD")
+	if dd == "" {
+		dd = "../../../fmsgd/dd.sql"
+	}
+	sql, err := os.ReadFile(dd)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err = pool.Exec(ctx, string(sql)); err != nil {
+		t.Fatal(err)
+	}
+	id := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `{"acceptingNew":true}`)
+	}))
+	t.Cleanup(id.Close)
+	h := NewMessageHandler(&db.DB{Pool: pool}, t.TempDir(), 1<<20, 2<<20, 256, nil, id.URL, "example.com")
+	a := NewAttachmentHandler(h.DB, h.DataDir, 1<<20, 2<<20)
+	gin.SetMode(gin.TestMode)
+	r := gin.New()
+	r.Use(func(c *gin.Context) { c.Set(middleware.IdentityKey, c.GetHeader("X-Test-Identity")) })
+	r.POST("/fmsg", h.Atomic((*MessageHandler).Create))
+	r.PUT("/fmsg/:id", h.Atomic((*MessageHandler).Update))
+	r.DELETE("/fmsg/:id", h.Atomic((*MessageHandler).Delete))
+	r.GET("/fmsg", h.List)
+	r.GET("/fmsg/sent", h.Sent)
+	r.GET("/fmsg/:id", h.Get)
+	r.GET("/fmsg/:id/data", h.DownloadData)
+	r.POST("/fmsg/:id/send", h.Atomic((*MessageHandler).Send))
+	r.POST("/fmsg/:id/read", h.MarkRead)
+	r.POST("/fmsg/:id/add-to", h.Atomic((*MessageHandler).AddRecipients))
+	r.POST("/fmsg/:id/react", h.Atomic((*MessageHandler).React))
+	r.GET("/fmsg/:id/thread", h.ThreadText)
+	r.GET("/fmsg/:id/thread/messages", h.ThreadMessages)
+	r.POST("/fmsg/:id/attach", a.Atomic((*AttachmentHandler).Upload))
+	r.GET("/fmsg/:id/attach/:filename", a.Download)
+	r.DELETE("/fmsg/:id/attach/:filename", a.Atomic((*AttachmentHandler).DeleteAttachment))
+	return &finalizationAPI{r, h, pool}
+}
+func (a *finalizationAPI) request(identity, method, path string, body any) *httptest.ResponseRecorder {
+	data, _ := json.Marshal(body)
+	req := httptest.NewRequest(method, path, bytes.NewReader(data))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Test-Identity", identity)
+	w := httptest.NewRecorder()
+	a.router.ServeHTTP(w, req)
+	return w
+}
+func jsonResponse(t *testing.T, w *httptest.ResponseRecorder, status int) map[string]any {
+	t.Helper()
+	if w.Code != status {
+		t.Fatalf("HTTP %d, want %d: %s", w.Code, status, w.Body.String())
+	}
+	var v map[string]any
+	if err := json.Unmarshal(w.Body.Bytes(), &v); err != nil {
+		t.Fatal(err)
+	}
+	return v
+}
+func (a *finalizationAPI) draft(t *testing.T, identity, body string, pid any) string {
+	t.Helper()
+	v := jsonResponse(t, a.request(identity, "POST", "/fmsg", map[string]any{"version": 1, "from": identity, "to": []string{"@bob@example.com", "@alice@example.com"}, "type": "text/plain;charset=UTF-8", "data": body, "pid": pid}), 201)
+	return fmt.Sprintf("%.0f", v["id"])
+}
+func (a *finalizationAPI) upload(t *testing.T, id string) {
+	t.Helper()
+	var b bytes.Buffer
+	m := multipart.NewWriter(&b)
+	f, err := m.CreateFormFile("file", "note.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _ = f.Write([]byte("attachment bytes"))
+	_ = m.Close()
+	r := httptest.NewRequest("POST", "/fmsg/"+id+"/attach", &b)
+	r.Header.Set("Content-Type", m.FormDataContentType())
+	r.Header.Set("X-Test-Identity", "@alice@example.com")
+	w := httptest.NewRecorder()
+	a.router.ServeHTTP(w, r)
+	if w.Code != 201 {
+		t.Fatalf("upload %d: %s", w.Code, w.Body)
+	}
+}
+
+func TestFinalizationLocalHashReferences(t *testing.T) {
+	a := newFinalizationAPI(t)
+	alice, bob := "@alice@example.com", "@bob@example.com"
+	id := a.draft(t, alice, strings.Repeat("compressible content ", 150), nil)
+	a.upload(t, id)
+	draft := jsonResponse(t, a.request(alice, "GET", "/fmsg/"+id, nil), 200)
+	if draft["sha256"] != nil {
+		t.Fatal("draft has hash")
+	}
+	sent := jsonResponse(t, a.request(alice, "POST", "/fmsg/"+id+"/send", nil), 200)
+	hash := sent["sha256"].(string)
+	if len(hash) != 64 {
+		t.Fatal(hash)
+	}
+	var snapshot, stored []byte
+	var stamp float64
+	if err := a.pool.QueryRow(context.Background(), `SELECT wire_message,sha256,time_sent FROM msg WHERE id=$1`, id).Scan(&snapshot, &stored, &stamp); err != nil {
+		t.Fatal(err)
+	}
+	wire, err := fmsg.UnmarshalPrepared(snapshot, stored)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if wire.Flags&fmsg.FlagDeflate == 0 {
+		t.Fatal("body was not compressed before hashing")
+	}
+	if stamp != sent["time"] {
+		t.Fatal("timestamp differs from hashed timestamp")
+	}
+	for _, suffix := range []string{"", "/data", "/attach/note.txt", "/thread", "/thread/messages"} {
+		byID := a.request(bob, "GET", "/fmsg/"+id+suffix, nil)
+		byHash := a.request(bob, "GET", "/fmsg/"+strings.ToUpper(hash)+suffix, nil)
+		if byID.Code != 200 || byHash.Code != 200 || !bytes.Equal(byID.Body.Bytes(), byHash.Body.Bytes()) {
+			t.Fatalf("reference mismatch for %s: %d %d %s", suffix, byID.Code, byHash.Code, byHash.Body)
+		}
+		denied := a.request("@outsider@example.com", "GET", "/fmsg/"+hash+suffix, nil)
+		if denied.Code != 403 {
+			t.Fatalf("unauthorized %s: %d", suffix, denied.Code)
+		}
+	}
+	for _, route := range []string{"/fmsg", "/fmsg/sent"} {
+		w := a.request(alice, "GET", route, nil)
+		if w.Code != 200 || !strings.Contains(w.Body.String(), hash) {
+			t.Fatalf("list missing hash: %s", w.Body)
+		}
+	}
+	item, err := a.h.messageItemFor(context.Background(), mustID(t, id), bob)
+	if err != nil || item.SHA256 == nil || *item.SHA256 != hash {
+		t.Fatalf("WebSocket payload missing hash: %+v %v", item, err)
+	}
+	if w := a.request(bob, "POST", "/fmsg/"+hash+"/read", nil); w.Code != 200 {
+		t.Fatal(w.Code, w.Body)
+	}
+	for _, path := range []string{strings.Repeat("g", 64), "0", "-1", "18446744073709551616"} {
+		if w := a.request(alice, "GET", "/fmsg/"+path, nil); w.Code != 400 {
+			t.Fatal(path, w.Code)
+		}
+	}
+	if w := a.request(alice, "GET", "/fmsg/"+strings.Repeat("1", 64), nil); w.Code != 404 {
+		t.Fatal("numeric-only hash", w.Code)
+	}
+	for _, parent := range []any{mustID(t, id), hash} {
+		reply := a.draft(t, bob, "local reply", parent)
+		jsonResponse(t, a.request(bob, "POST", "/fmsg/"+reply+"/send", nil), 200)
+		got := jsonResponse(t, a.request(bob, "GET", "/fmsg/"+reply, nil), 200)
+		if got["psha256"] != hash {
+			t.Fatal(got)
+		}
+	}
+	batch := jsonResponse(t, a.request(alice, "POST", "/fmsg/"+hash+"/add-to", map[string]any{"add_to": []string{"@carol@example.com"}}), 200)
+	batchHash := batch["sha256"].(string)
+	if len(batchHash) != 64 || batchHash == hash {
+		t.Fatal(batch)
+	}
+	reply := a.draft(t, "@carol@example.com", "reply to batch", batchHash)
+	jsonResponse(t, a.request("@carol@example.com", "POST", "/fmsg/"+reply+"/send", nil), 200)
+	got := jsonResponse(t, a.request("@carol@example.com", "GET", "/fmsg/"+reply, nil), 200)
+	if got["psha256"] != batchHash {
+		t.Fatal(got)
+	}
+	reaction := jsonResponse(t, a.request(bob, "POST", "/fmsg/"+hash+"/react", map[string]any{"emoji": "👍"}), 201)
+	if len(reaction["sha256"].(string)) != 64 {
+		t.Fatal(reaction)
+	}
+	idempotent := jsonResponse(t, a.request(bob, "POST", "/fmsg/"+hash+"/react", map[string]any{"emoji": "👍"}), 200)
+	if reaction["sha256"] != idempotent["sha256"] {
+		t.Fatal("reaction changed")
+	}
+	if w := a.request(alice, "DELETE", "/fmsg/"+hash+"/attach/note.txt", nil); w.Code != 403 {
+		t.Fatal(w.Code)
+	}
+	if w := a.request(alice, "PUT", "/fmsg/"+hash, map[string]any{}); w.Code != 403 {
+		t.Fatal(w.Code)
+	}
+	if w := a.request(alice, "POST", "/fmsg/"+hash+"/send", nil); w.Code != 409 {
+		t.Fatal(w.Code)
+	}
+}
+func mustID(t *testing.T, id string) int64 {
+	t.Helper()
+	var n int64
+	if _, err := fmt.Sscan(id, &n); err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
+func TestFinalizationRollbackAndConcurrentSend(t *testing.T) {
+	a := newFinalizationAPI(t)
+	alice := "@alice@example.com"
+	id := a.draft(t, alice, "missing payload", nil)
+	var path string
+	if err := a.pool.QueryRow(context.Background(), `SELECT filepath FROM msg WHERE id=$1`, id).Scan(&path); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Remove(path); err != nil {
+		t.Fatal(err)
+	}
+	if w := a.request(alice, "POST", "/fmsg/"+id+"/send", nil); w.Code != 500 {
+		t.Fatal(w.Code, w.Body)
+	}
+	got := jsonResponse(t, a.request(alice, "GET", "/fmsg/"+id, nil), 200)
+	if got["time"] != nil || got["sha256"] != nil {
+		t.Fatal("failed finalization committed", got)
+	}
+	files, _ := filepath.Glob(filepath.Join(filepath.Dir(path), ".fmsg-wire-*"))
+	if len(files) > 0 {
+		t.Fatal("failed finalization leaked files", files)
+	}
+	id = a.draft(t, alice, "race", nil)
+	var wg sync.WaitGroup
+	codes := make(chan int, 8)
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() { defer wg.Done(); codes <- a.request(alice, "POST", "/fmsg/"+id+"/send", nil).Code }()
+	}
+	wg.Wait()
+	close(codes)
+	ok := 0
+	for c := range codes {
+		if c == 200 {
+			ok++
+		} else if c != 409 {
+			t.Fatal(c)
+		}
+	}
+	if ok != 1 {
+		t.Fatal("successful sends", ok)
+	}
+	// Race a full draft edit with send. Either order is valid; the committed
+	// digest must describe exactly the content that remains downloadable.
+	for i := 0; i < 5; i++ {
+		id = a.draft(t, alice, "before", nil)
+		wg.Add(2)
+		go func(id string) { defer wg.Done(); a.request(alice, "POST", "/fmsg/"+id+"/send", nil) }(id)
+		go func(id string) {
+			defer wg.Done()
+			a.request(alice, "PUT", "/fmsg/"+id, map[string]any{"version": 1, "from": alice, "to": []string{"@bob@example.com"}, "type": "text/plain", "data": "after"})
+		}(id)
+		wg.Wait()
+		var snapshot, hash []byte
+		if err := a.pool.QueryRow(context.Background(), `SELECT wire_message,sha256 FROM msg WHERE id=$1`, id).Scan(&snapshot, &hash); err != nil {
+			t.Fatal(err)
+		}
+		h, err := fmsg.UnmarshalPrepared(snapshot, hash)
+		if err != nil {
+			t.Fatal(err)
+		}
+		wire, _ := os.ReadFile(h.Filepath)
+		body := a.request(alice, "GET", "/fmsg/"+hex.EncodeToString(hash)+"/data", nil)
+		if body.Code != 200 || !bytes.Equal(wire, body.Body.Bytes()) {
+			t.Fatal("content changed after finalization")
+		}
+	}
+}
+
+func TestFinalizationCommitFailureDoesNotReturnSuccess(t *testing.T) {
+	a := newFinalizationAPI(t)
+	id := a.draft(t, "@alice@example.com", "commit failure", nil)
+	_, err := a.pool.Exec(context.Background(), `CREATE FUNCTION reject_test_send() RETURNS trigger AS $$ BEGIN
+	 IF NEW.time_sent IS NOT NULL THEN RAISE EXCEPTION 'test commit rejection'; END IF; RETURN NEW; END; $$ LANGUAGE plpgsql;
+	 CREATE CONSTRAINT TRIGGER reject_test_send AFTER UPDATE ON msg DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION reject_test_send()`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	response := a.request("@alice@example.com", "POST", "/fmsg/"+id+"/send", nil)
+	if response.Code != 500 || strings.Contains(response.Body.String(), "sha256") {
+		t.Fatalf("success escaped before commit: %d %s", response.Code, response.Body)
+	}
+	got := jsonResponse(t, a.request("@alice@example.com", "GET", "/fmsg/"+id, nil), 200)
+	if got["time"] != nil || got["sha256"] != nil {
+		t.Fatal("rejected commit changed draft", got)
+	}
+}
diff --git a/internal/handlers/messages.go b/internal/handlers/messages.go
index a045446..c660add 100644
--- a/internal/handlers/messages.go
+++ b/internal/handlers/messages.go
@@ -3,8 +3,11 @@ package handlers
 
 import (
 	"context"
+	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/markmnl/fmsgd/pkg/message"
 	"io"
 	"log"
 	"mime"
@@ -27,6 +30,7 @@ import (
 
 // MessageHandler holds dependencies for message routes.
 type MessageHandler struct {
+	query         queries
 	DB            *db.DB
 	DataDir       string
 	MaxDataSize   int64
@@ -61,6 +65,8 @@ func (h *MessageHandler) visibleAddrs(c *gin.Context) ([]string, error) {
 // messageListItem is the JSON shape for each message in the list response.
 // It mirrors the single-message response (including an id).
 type messageListItem struct {
+	SHA256      *string                    `json:"sha256"`
+	PSHA256     *string                    `json:"psha256"`
 	ID          int64                      `json:"id"`
 	Version     int                        `json:"version"`
 	HasPid      bool                       `json:"has_pid"`
@@ -138,7 +144,7 @@ func (h *MessageHandler) resolveLocalDelivery(ctx context.Context, table string,
 			log.Printf("resolve local delivery: unexpected fmsgid status %d for %s", code, addr)
 			continue
 		}
-		if _, err := h.DB.Pool.Exec(ctx, query, delivered, responseCode, msgID, addr); err != nil {
+		if _, err := h.q().Exec(ctx, query, delivered, responseCode, msgID, addr); err != nil {
 			log.Printf("resolve local delivery: update %s msg %d addr %s: %v", table, msgID, addr, err)
 		}
 	}
@@ -223,10 +229,10 @@ func remoteRecipientDomains(msg *models.Message, localDomain string) []string {
 // summary of its recorded recipient delivery outcomes.
 func (h *MessageHandler) parentDeliveryByDomain(ctx context.Context, parentID int64) (string, map[string]parentDomainDelivery, error) {
 	var fromAddr string
-	if err := h.DB.Pool.QueryRow(ctx, "SELECT from_addr FROM msg WHERE id = $1", parentID).Scan(&fromAddr); err != nil {
+	if err := h.q().QueryRow(ctx, "SELECT from_addr FROM msg WHERE id = $1", parentID).Scan(&fromAddr); err != nil {
 		return "", nil, err
 	}
-	rows, err := h.DB.Pool.Query(ctx, `
+	rows, err := h.q().Query(ctx, `
 		SELECT addr, time_delivered IS NOT NULL, response_code FROM (
 			SELECT addr, time_delivered, response_code FROM msg_to WHERE msg_id = $1
 			UNION ALL
@@ -272,8 +278,9 @@ func (h *MessageHandler) parentDeliveryByDomain(ctx context.Context, parentID in
 // create/update, so add_to here is intentionally discarded.
 type messageInput struct {
 	models.Message
-	AddTo any    `json:"add_to"`
-	Data  string `json:"data"`
+	Parent json.RawMessage `json:"pid"`
+	AddTo  any             `json:"add_to"`
+	Data   string          `json:"data"`
 }
 
 // List handles GET /fmsg — lists messages where the authenticated user is a recipient.
@@ -292,8 +299,8 @@ func (h *MessageHandler) List(c *gin.Context) {
 
 	ctx := c.Request.Context()
 
-	rows, err := h.DB.Pool.Query(ctx,
-		`SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, m.from_addr, m.topic, m.type, m.size, m.filepath,
+	rows, err := h.q().Query(ctx,
+		`SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, encode(m.sha256,'hex'), encode(m.psha256,'hex'), m.from_addr, m.topic, m.type, m.size, m.filepath,
 		        COALESCE(
 		            (SELECT mt.time_read FROM msg_to mt WHERE mt.msg_id = m.id AND lower(mt.addr) = ANY($1)),
 		            (SELECT mat.time_read FROM msg_add_to mat WHERE mat.msg_id = m.id AND lower(mat.addr) = ANY($1))
@@ -317,7 +324,7 @@ func (h *MessageHandler) List(c *gin.Context) {
 	for rows.Next() {
 		var m messageListItem
 		var dataPath string
-		if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath, &m.TimeRead); err != nil {
+		if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.SHA256, &m.PSHA256, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath, &m.TimeRead); err != nil {
 			log.Printf("list messages scan: %v", err)
 			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list messages"})
 			return
@@ -350,7 +357,7 @@ func (h *MessageHandler) List(c *gin.Context) {
 	}
 
 	// Batch-load attachments.
-	attRows, err := h.DB.Pool.Query(ctx,
+	attRows, err := h.q().Query(ctx,
 		"SELECT msg_id, filename, filesize FROM msg_attachment WHERE msg_id = ANY($1)",
 		msgIDs,
 	)
@@ -396,8 +403,8 @@ func (h *MessageHandler) Sent(c *gin.Context) {
 
 	ctx := c.Request.Context()
 
-	rows, err := h.DB.Pool.Query(ctx,
-		`SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, m.from_addr, m.topic, m.type, m.size, m.filepath
+	rows, err := h.q().Query(ctx,
+		`SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, encode(m.sha256,'hex'), encode(m.psha256,'hex'), m.from_addr, m.topic, m.type, m.size, m.filepath
 		 FROM msg m
 		 WHERE lower(m.from_addr) = ANY($1)
 		 ORDER BY m.id DESC
@@ -416,7 +423,7 @@ func (h *MessageHandler) Sent(c *gin.Context) {
 	for rows.Next() {
 		var m messageListItem
 		var dataPath string
-		if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath); err != nil {
+		if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.SHA256, &m.PSHA256, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath); err != nil {
 			log.Printf("list sent messages scan: %v", err)
 			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list sent messages"})
 			return
@@ -448,7 +455,7 @@ func (h *MessageHandler) Sent(c *gin.Context) {
 	}
 
 	// Batch-load attachments.
-	attRows, err := h.DB.Pool.Query(ctx,
+	attRows, err := h.q().Query(ctx,
 		"SELECT msg_id, filename, filesize FROM msg_attachment WHERE msg_id = ANY($1)",
 		msgIDs,
 	)
@@ -498,6 +505,9 @@ func (h *MessageHandler) Create(c *gin.Context) {
 		return
 	}
 
+	if !h.resolveParent(c, &msg) {
+		return
+	}
 	if err := validatePidRelations(msg.PID, msg.Topic); err != nil {
 		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 		return
@@ -516,7 +526,7 @@ func (h *MessageHandler) Create(c *gin.Context) {
 	}
 
 	// Detect zip (deflate) content by checking for the zip magic bytes.
-	msg.Deflate = isZip([]byte(msg.Data))
+	msg.Deflate = false // wire compression is chosen only at finalization
 
 	// Parse extension from MIME type.
 	ext := mimeToExt(msg.Type)
@@ -524,11 +534,11 @@ func (h *MessageHandler) Create(c *gin.Context) {
 	// Insert message row with empty filepath; update after we know the ID.
 	dataSize := len(msg.Data)
 	var msgID int64
-	err := h.DB.Pool.QueryRow(ctx,
-		`INSERT INTO msg (version, pid, no_reply, is_important, is_deflate, is_terminal, from_addr, topic, type, size, filepath, time_sent)
- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, '', NULL)
+	err := h.q().QueryRow(ctx,
+		`INSERT INTO msg (version, pid, psha256, no_reply, is_important, is_deflate, is_terminal, from_addr, topic, type, size, filepath, time_sent)
+ VALUES ($1, $2, decode($11,'hex'), $3, $4, $5, $6, $7, $8, $9, $10, '', NULL)
  RETURNING id`,
-		msg.Version, msg.PID, msg.NoReply, msg.Important, msg.Deflate, msg.Terminal, msg.From, msg.Topic, msg.Type, dataSize,
+		msg.Version, msg.PID, msg.NoReply, msg.Important, msg.Deflate, msg.Terminal, msg.From, msg.Topic, msg.Type, dataSize, msg.PSHA256,
 	).Scan(&msgID)
 	if err != nil {
 		log.Printf("create message: insert: %v", err)
@@ -541,23 +551,28 @@ func (h *MessageHandler) Create(c *gin.Context) {
 	if err != nil {
 		log.Printf("create message: save data: %v", err)
 		// Attempt rollback.
-		_, _ = h.DB.Pool.Exec(ctx, "DELETE FROM msg WHERE id = $1", msgID)
+		_, _ = h.q().Exec(ctx, "DELETE FROM msg WHERE id = $1", msgID)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to save message data"})
 		return
 	}
 
+	createdFile(c, dataPath)
 	// Update filepath in the database.
-	if _, err = h.DB.Pool.Exec(ctx, "UPDATE msg SET filepath = $1 WHERE id = $2", dataPath, msgID); err != nil {
+	if _, err = h.q().Exec(ctx, "UPDATE msg SET filepath = $1 WHERE id = $2", dataPath, msgID); err != nil {
 		log.Printf("create message: update filepath: %v", err)
+		c.JSON(500, gin.H{"error": "failed to save message"})
+		return
 	}
 
 	// Insert recipients.
 	for _, addr := range msg.To {
-		if _, err = h.DB.Pool.Exec(ctx,
+		if _, err = h.q().Exec(ctx,
 			"INSERT INTO msg_to (msg_id, addr) VALUES ($1, $2) ON CONFLICT DO NOTHING",
 			msgID, addr,
 		); err != nil {
 			log.Printf("create message: insert recipient %s: %v", addr, err)
+			c.JSON(500, gin.H{"error": "failed to save recipients"})
+			return
 		}
 	}
 
@@ -572,7 +587,7 @@ func (h *MessageHandler) Get(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve message"})
 		return
 	}
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -608,7 +623,7 @@ func (h *MessageHandler) Get(c *gin.Context) {
 	// has no read state of their own.
 	if !isSender {
 		var timeRead *float64
-		err := h.DB.Pool.QueryRow(ctx,
+		err := h.q().QueryRow(ctx,
 			`SELECT COALESCE(
 			    (SELECT mt.time_read FROM msg_to mt WHERE mt.msg_id = $1 AND lower(mt.addr) = ANY($2)),
 			    (SELECT mat.time_read FROM msg_add_to mat WHERE mat.msg_id = $1 AND lower(mat.addr) = ANY($2))
@@ -632,7 +647,7 @@ func (h *MessageHandler) DownloadData(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve message"})
 		return
 	}
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -642,7 +657,7 @@ func (h *MessageHandler) DownloadData(c *gin.Context) {
 	// Fetch message metadata for auth check and file path.
 	var fromAddr string
 	var dataPath string
-	err = h.DB.Pool.QueryRow(ctx,
+	err = h.q().QueryRow(ctx,
 		"SELECT from_addr, filepath FROM msg WHERE id = $1", msgID,
 	).Scan(&fromAddr, &dataPath)
 	if err != nil {
@@ -658,7 +673,7 @@ func (h *MessageHandler) DownloadData(c *gin.Context) {
 	// Authorize: must be owner or recipient (across all visible addrs).
 	if !isRecipient(addrs, fromAddr) {
 		var recipientCount int
-		if err = h.DB.Pool.QueryRow(ctx,
+		if err = h.q().QueryRow(ctx,
 			`SELECT COUNT(*) FROM (
 				SELECT 1 FROM msg_to WHERE msg_id = $1 AND lower(addr) = ANY($2)
 				UNION ALL
@@ -689,13 +704,13 @@ func (h *MessageHandler) DownloadData(c *gin.Context) {
 // Update handles PUT /fmsg/:id — updates a draft message.
 func (h *MessageHandler) Update(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
 
 	ctx := c.Request.Context()
-	existing, _, err := h.fetchMessage(ctx, msgID)
+	existing, oldDataPath, err := h.fetchMessage(ctx, msgID)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
 			c.JSON(http.StatusNotFound, gin.H{"error": "message not found"})
@@ -730,6 +745,9 @@ func (h *MessageHandler) Update(c *gin.Context) {
 		return
 	}
 
+	if !h.resolveParent(c, &msg) {
+		return
+	}
 	if err := validatePidRelations(msg.PID, msg.Topic); err != nil {
 		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 		return
@@ -746,7 +764,7 @@ func (h *MessageHandler) Update(c *gin.Context) {
 
 	// Check total message size (data + existing attachments).
 	var attachTotal int64
-	if err := h.DB.Pool.QueryRow(ctx,
+	if err := h.q().QueryRow(ctx,
 		"SELECT COALESCE(SUM(filesize), 0) FROM msg_attachment WHERE msg_id = $1",
 		msgID,
 	).Scan(&attachTotal); err != nil {
@@ -759,7 +777,7 @@ func (h *MessageHandler) Update(c *gin.Context) {
 		return
 	}
 
-	msg.Deflate = isZip([]byte(msg.Data))
+	msg.Deflate = false // wire compression is chosen only at finalization
 	ext := mimeToExt(msg.Type)
 
 	dataPath, err := h.saveMessageData(msg.From, msgID, ext, msg.Data)
@@ -769,9 +787,10 @@ func (h *MessageHandler) Update(c *gin.Context) {
 		return
 	}
 
-	_, err = h.DB.Pool.Exec(ctx,
-		`UPDATE msg SET version=$1, pid=$2, no_reply=$3, is_important=$4, is_deflate=$5, is_terminal=$6, topic=$7, type=$8, size=$9, filepath=$10 WHERE id=$11`,
-		msg.Version, msg.PID, msg.NoReply, msg.Important, msg.Deflate, msg.Terminal, msg.Topic, msg.Type, len(msg.Data), dataPath, msgID,
+	createdFile(c, dataPath)
+	_, err = h.q().Exec(ctx,
+		`UPDATE msg SET version=$1, pid=$2, no_reply=$3, is_important=$4, is_deflate=$5, is_terminal=$6, topic=$7, type=$8, size=$9, filepath=$10, psha256=decode($12,'hex') WHERE id=$11`,
+		msg.Version, msg.PID, msg.NoReply, msg.Important, msg.Deflate, msg.Terminal, msg.Topic, msg.Type, len(msg.Data), dataPath, msgID, msg.PSHA256,
 	)
 	if err != nil {
 		log.Printf("update message %d: %v", msgID, err)
@@ -780,25 +799,34 @@ func (h *MessageHandler) Update(c *gin.Context) {
 	}
 
 	// Replace recipients.
-	if _, err = h.DB.Pool.Exec(ctx, "DELETE FROM msg_to WHERE msg_id = $1", msgID); err != nil {
+	if _, err = h.q().Exec(ctx, "DELETE FROM msg_to WHERE msg_id = $1", msgID); err != nil {
 		log.Printf("update message %d delete recipients: %v", msgID, err)
+		c.JSON(500, gin.H{"error": "failed to update recipients"})
+		return
 	}
 	for _, addr := range msg.To {
-		if _, err = h.DB.Pool.Exec(ctx,
+		if _, err = h.q().Exec(ctx,
 			"INSERT INTO msg_to (msg_id, addr) VALUES ($1, $2) ON CONFLICT DO NOTHING",
 			msgID, addr,
 		); err != nil {
 			log.Printf("update message %d insert recipient %s: %v", msgID, addr, err)
+			c.JSON(500, gin.H{"error": "failed to update recipients"})
+			return
 		}
 	}
 
+	afterCommit(c, func() {
+		if oldDataPath != "" {
+			_ = os.Remove(oldDataPath)
+		}
+	})
 	c.JSON(http.StatusOK, gin.H{"id": msgID})
 }
 
 // Delete handles DELETE /fmsg/:id — deletes a draft message.
 func (h *MessageHandler) Delete(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -825,7 +853,7 @@ func (h *MessageHandler) Delete(c *gin.Context) {
 	}
 
 	// Remove attachment files from disk.
-	rows, err := h.DB.Pool.Query(ctx, "SELECT filepath FROM msg_attachment WHERE msg_id = $1", msgID)
+	rows, err := h.q().Query(ctx, "SELECT filepath FROM msg_attachment WHERE msg_id = $1", msgID)
 	if err == nil {
 		var paths []string
 		for rows.Next() {
@@ -836,32 +864,40 @@ func (h *MessageHandler) Delete(c *gin.Context) {
 		}
 		rows.Close()
 		for _, p := range paths {
-			_ = os.Remove(p)
+			afterCommit(c, func() { _ = os.Remove(p) })
 		}
 	}
 
-	if _, err = h.DB.Pool.Exec(ctx, "DELETE FROM msg_attachment WHERE msg_id = $1", msgID); err != nil {
+	if _, err = h.q().Exec(ctx, "DELETE FROM msg_attachment WHERE msg_id = $1", msgID); err != nil {
 		log.Printf("delete message %d: delete attachments: %v", msgID, err)
 	}
-	if _, err = h.DB.Pool.Exec(ctx, "DELETE FROM msg_to WHERE msg_id = $1", msgID); err != nil {
+	if _, err = h.q().Exec(ctx, "DELETE FROM msg_to WHERE msg_id = $1", msgID); err != nil {
 		log.Printf("delete message %d: delete recipients: %v", msgID, err)
 	}
-	if _, err = h.DB.Pool.Exec(ctx, "DELETE FROM msg_add_to WHERE msg_id = $1", msgID); err != nil {
+	if _, err = h.q().Exec(ctx, "DELETE FROM msg_add_to WHERE msg_id = $1", msgID); err != nil {
 		log.Printf("delete message %d: delete add_to recipients: %v", msgID, err)
 	}
 
+	if _, err = h.q().Exec(ctx, `DELETE FROM msg_add_to_notify WHERE batch_id IN (SELECT id FROM msg_add_to_batch WHERE msg_id=$1)`, msgID); err != nil {
+		c.JSON(500, gin.H{"error": "failed to delete notifications"})
+		return
+	}
+	if _, err = h.q().Exec(ctx, `DELETE FROM msg_add_to_batch WHERE msg_id=$1`, msgID); err != nil {
+		c.JSON(500, gin.H{"error": "failed to delete batches"})
+		return
+	}
 	// Get data filepath before deleting.
 	var dataPath string
-	_ = h.DB.Pool.QueryRow(ctx, "SELECT filepath FROM msg WHERE id = $1", msgID).Scan(&dataPath)
+	_ = h.q().QueryRow(ctx, "SELECT filepath FROM msg WHERE id = $1", msgID).Scan(&dataPath)
 
-	if _, err = h.DB.Pool.Exec(ctx, "DELETE FROM msg WHERE id = $1", msgID); err != nil {
+	if _, err = h.q().Exec(ctx, "DELETE FROM msg WHERE id = $1", msgID); err != nil {
 		log.Printf("delete message %d: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"})
 		return
 	}
 
 	if dataPath != "" {
-		_ = os.Remove(dataPath)
+		afterCommit(c, func() { _ = os.Remove(dataPath) })
 	}
 
 	c.Status(http.StatusNoContent)
@@ -870,7 +906,7 @@ func (h *MessageHandler) Delete(c *gin.Context) {
 // Send handles POST /fmsg/:id/send — marks a message as sent.
 func (h *MessageHandler) Send(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -926,20 +962,31 @@ func (h *MessageHandler) Send(c *gin.Context) {
 	}
 
 	now := float64(time.Now().UnixMicro()) / 1e6
-	if _, err = h.DB.Pool.Exec(ctx, "UPDATE msg SET time_sent = $1 WHERE id = $2", now, msgID); err != nil {
-		log.Printf("send message %d: %v", msgID, err)
-		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to send message"})
+	tx, ok := h.query.(pgx.Tx)
+	if !ok {
+		c.JSON(500, gin.H{"error": "send requires a message transaction"})
+		return
+	}
+	var files message.Files
+	onRollback(c, func() { files.Cleanup() })
+	hash, err := message.Finalize(ctx, finalizationTx{tx}, msgID, now, &files)
+	if err != nil {
+		log.Printf("finalize message %d: %v", msgID, err)
+		c.JSON(500, gin.H{"error": "failed to finalize message"})
 		return
 	}
 
 	// fmsgd's outbound sender skips the local domain entirely, so local
 	// recipients need their delivery status resolved here instead.
-	h.resolveLocalDelivery(ctx, "msg_to", msgID, h.LocalDomain, existing.To)
-	for _, b := range existing.AddTo {
-		h.resolveLocalDelivery(ctx, "msg_add_to", msgID, h.LocalDomain, b.To)
-	}
-
-	c.JSON(http.StatusOK, gin.H{"id": msgID, "time": now})
+	afterCommit(c, func() {
+		plain := *h
+		plain.query = nil
+		plain.resolveLocalDelivery(ctx, "msg_to", msgID, h.LocalDomain, existing.To)
+		for _, b := range existing.AddTo {
+			plain.resolveLocalDelivery(ctx, "msg_add_to", msgID, h.LocalDomain, b.To)
+		}
+	})
+	c.JSON(http.StatusOK, gin.H{"id": msgID, "time": now, "sha256": hex.EncodeToString(hash)})
 }
 
 // MarkRead handles POST /fmsg/:id/read — marks a message as read by the
@@ -947,7 +994,7 @@ func (h *MessageHandler) Send(c *gin.Context) {
 // the original time_read without updating it.
 func (h *MessageHandler) MarkRead(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -960,7 +1007,7 @@ func (h *MessageHandler) MarkRead(c *gin.Context) {
 	// an EXISTS check.
 	var existing *float64
 	var recipient bool
-	err := h.DB.Pool.QueryRow(ctx,
+	err := h.q().QueryRow(ctx,
 		`SELECT
 		    COALESCE(
 		        (SELECT mt.time_read FROM msg_to mt WHERE mt.msg_id = $1 AND lower(mt.addr) = lower($2)),
@@ -989,7 +1036,7 @@ func (h *MessageHandler) MarkRead(c *gin.Context) {
 	// Update whichever recipient row matches; only one of these will affect
 	// rows for any given (msg_id, addr) pair given the unique constraint on
 	// each table.
-	if _, err = h.DB.Pool.Exec(ctx,
+	if _, err = h.q().Exec(ctx,
 		`UPDATE msg_to SET time_read = $1
 		 WHERE msg_id = $2 AND lower(addr) = lower($3) AND time_read IS NULL`,
 		now, msgID, identity,
@@ -998,7 +1045,7 @@ func (h *MessageHandler) MarkRead(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to mark message read"})
 		return
 	}
-	if _, err = h.DB.Pool.Exec(ctx,
+	if _, err = h.q().Exec(ctx,
 		`UPDATE msg_add_to SET time_read = $1
 		 WHERE msg_id = $2 AND lower(addr) = lower($3) AND time_read IS NULL`,
 		now, msgID, identity,
@@ -1019,7 +1066,7 @@ type addToInput struct {
 // AddRecipients handles POST /fmsg/:id/add-to — adds additional recipients to a message.
 func (h *MessageHandler) AddRecipients(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -1050,7 +1097,7 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 	var fromAddr string
 	var timeSent *float64
 	var terminal bool
-	err := h.DB.Pool.QueryRow(ctx,
+	err := h.q().QueryRow(ctx,
 		"SELECT from_addr, time_sent, is_terminal FROM msg WHERE id = $1", msgID,
 	).Scan(&fromAddr, &timeSent, &terminal)
 	if err != nil {
@@ -1072,7 +1119,7 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 	// Verify the requester is an existing participant (from or msg_to).
 	if !sameAddr(fromAddr, identity) {
 		var recipientCount int
-		if err = h.DB.Pool.QueryRow(ctx,
+		if err = h.q().QueryRow(ctx,
 			"SELECT COUNT(*) FROM msg_to WHERE msg_id = $1 AND lower(addr) = lower($2)", msgID, identity,
 		).Scan(&recipientCount); err != nil || recipientCount == 0 {
 			c.JSON(http.StatusForbidden, gin.H{"error": "only existing participants may add recipients"})
@@ -1097,7 +1144,7 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 		loweredAddTo[i] = strings.ToLower(addr)
 	}
 	var alreadyAdded int
-	if err = h.DB.Pool.QueryRow(ctx,
+	if err = h.q().QueryRow(ctx,
 		"SELECT COUNT(*) FROM msg_add_to WHERE msg_id = $1 AND lower(addr) = ANY($2)",
 		msgID, loweredAddTo,
 	).Scan(&alreadyAdded); err != nil {
@@ -1112,7 +1159,7 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 
 	// Insert the new add_to recipients and record who added them. Both run in a
 	// single transaction so a partial failure leaves the message unchanged.
-	tx, err := h.DB.Pool.Begin(ctx)
+	tx, err := h.q().Begin(ctx)
 	if err != nil {
 		log.Printf("add recipients: begin tx for msg %d: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add recipients"})
@@ -1172,6 +1219,12 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 		return
 	}
 
+	hash, err := message.FinalizeBatch(ctx, finalizationTx{tx}, msgID, batchID)
+	if err != nil {
+		log.Printf("finalize batch %d: %v", batchID, err)
+		c.JSON(500, gin.H{"error": "failed to finalize add-to batch"})
+		return
+	}
 	if err = tx.Commit(ctx); err != nil {
 		log.Printf("add recipients: commit tx for msg %d: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add recipients"})
@@ -1185,10 +1238,19 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 	// whoever called this endpoint — the caller adding recipients may be a
 	// federated participant on a different domain than the recipients they add.
 	if timeSent != nil {
-		h.resolveLocalDelivery(ctx, "msg_add_to", msgID, h.LocalDomain, input.AddTo)
+		afterCommit(c, func() {
+			plain := *h
+			plain.query = nil
+			plain.resolveLocalDelivery(ctx, "msg_add_to", msgID, h.LocalDomain, input.AddTo)
+		})
+	}
+
+	var encoded any
+	if len(hash) > 0 {
+		encoded = hex.EncodeToString(hash)
 	}
+	c.JSON(http.StatusOK, gin.H{"id": msgID, "added": len(input.AddTo), "batch_id": batchID, "sha256": encoded})
 
-	c.JSON(http.StatusOK, gin.H{"id": msgID, "added": len(input.AddTo)})
 }
 
 // loadRecipients loads the direct (msg_to) recipients for the given messages,
@@ -1196,7 +1258,7 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 func (h *MessageHandler) loadRecipients(ctx context.Context, msgIDs []int64) (map[int64][]string, map[int64][]models.RecipientDelivery) {
 	toMap := make(map[int64][]string)
 	deliveryMap := make(map[int64][]models.RecipientDelivery)
-	rows, err := h.DB.Pool.Query(ctx,
+	rows, err := h.q().Query(ctx,
 		"SELECT msg_id, addr, time_delivered, response_code FROM msg_to WHERE msg_id = ANY($1)",
 		msgIDs,
 	)
@@ -1242,8 +1304,8 @@ func int16PtrToIntPtr(v *int16) *int {
 // their insertion order (batch id).
 func (h *MessageHandler) loadAddToBatches(ctx context.Context, msgIDs []int64) map[int64][]models.AddToBatch {
 	result := make(map[int64][]models.AddToBatch)
-	rows, err := h.DB.Pool.Query(ctx,
-		`SELECT b.msg_id, b.id, b.add_to_from, b.time_added, mat.addr, mat.time_delivered, mat.response_code
+	rows, err := h.q().Query(ctx,
+		`SELECT b.msg_id, b.id, b.add_to_from, b.time_added, encode(b.sha256,'hex'), mat.addr, mat.time_delivered, mat.response_code
 		 FROM msg_add_to_batch b
 		 LEFT JOIN msg_add_to mat ON mat.batch_id = b.id
 		 WHERE b.msg_id = ANY($1)
@@ -1262,10 +1324,11 @@ func (h *MessageHandler) loadAddToBatches(ctx context.Context, msgIDs []int64) m
 		var msgID, batchID int64
 		var addToFrom string
 		var timeAdded float64
+		var hash *string
 		var addr *string
 		var timeDelivered *float64
 		var responseCode *int16
-		if err := rows.Scan(&msgID, &batchID, &addToFrom, &timeAdded, &addr, &timeDelivered, &responseCode); err != nil {
+		if err := rows.Scan(&msgID, &batchID, &addToFrom, &timeAdded, &hash, &addr, &timeDelivered, &responseCode); err != nil {
 			log.Printf("load add_to batches scan: %v", err)
 			continue
 		}
@@ -1273,6 +1336,7 @@ func (h *MessageHandler) loadAddToBatches(ctx context.Context, msgIDs []int64) m
 		if !ok {
 			result[msgID] = append(result[msgID], models.AddToBatch{
 				BatchID:   batchID,
+				SHA256:    hash,
 				AddToFrom: addToFrom,
 				Time:      timeAdded,
 			})
@@ -1295,8 +1359,8 @@ func (h *MessageHandler) loadAddToBatches(ctx context.Context, msgIDs []int64) m
 // It also returns the raw filepath stored in the database so callers can use it
 // after performing their own authorization checks.
 func (h *MessageHandler) fetchMessage(ctx context.Context, msgID int64) (*models.Message, string, error) {
-	row := h.DB.Pool.QueryRow(ctx,
-		`SELECT version, pid, no_reply, is_important, is_deflate, is_terminal, time_sent, from_addr, topic, type, size, filepath FROM msg WHERE id = $1`,
+	row := h.q().QueryRow(ctx,
+		`SELECT version, pid, no_reply, is_important, is_deflate, is_terminal, time_sent, encode(sha256,'hex'), encode(psha256,'hex'), from_addr, topic, type, size, filepath FROM msg WHERE id = $1`,
 		msgID,
 	)
 
@@ -1304,7 +1368,7 @@ func (h *MessageHandler) fetchMessage(ctx context.Context, msgID int64) (*models
 	var pid *int64
 	var timeSent *float64
 	var dataPath string
-	if err := row.Scan(&msg.Version, &pid, &msg.NoReply, &msg.Important, &msg.Deflate, &msg.Terminal, &timeSent, &msg.From, &msg.Topic, &msg.Type, &msg.Size, &dataPath); err != nil {
+	if err := row.Scan(&msg.Version, &pid, &msg.NoReply, &msg.Important, &msg.Deflate, &msg.Terminal, &timeSent, &msg.SHA256, &msg.PSHA256, &msg.From, &msg.Topic, &msg.Type, &msg.Size, &dataPath); err != nil {
 		return nil, "", err
 	}
 	msg.PID = pid
@@ -1321,7 +1385,7 @@ func (h *MessageHandler) fetchMessage(ctx context.Context, msgID int64) (*models
 	msg.HasAddTo = len(msg.AddTo) > 0
 
 	// Load attachments.
-	attRows, err := h.DB.Pool.Query(ctx, "SELECT filename, filesize FROM msg_attachment WHERE msg_id = $1", msgID)
+	attRows, err := h.q().Query(ctx, "SELECT filename, filesize FROM msg_attachment WHERE msg_id = $1", msgID)
 	if err == nil {
 		for attRows.Next() {
 			var a models.Attachment
@@ -1347,6 +1411,8 @@ func (h *MessageHandler) messageItemFor(ctx context.Context, msgID int64, recipi
 
 	item := &messageListItem{
 		ID:          msgID,
+		SHA256:      msg.SHA256,
+		PSHA256:     msg.PSHA256,
 		Version:     msg.Version,
 		HasPid:      msg.HasPid,
 		HasAddTo:    msg.HasAddTo,
@@ -1375,7 +1441,7 @@ func (h *MessageHandler) messageItemFor(ctx context.Context, msgID int64, recipi
 
 	// Populate the recipient's per-recipient read state, mirroring Get.
 	var timeRead *float64
-	if err := h.DB.Pool.QueryRow(ctx,
+	if err := h.q().QueryRow(ctx,
 		`SELECT COALESCE(
 		    (SELECT mt.time_read FROM msg_to mt WHERE mt.msg_id = $1 AND lower(mt.addr) = lower($2)),
 		    (SELECT mat.time_read FROM msg_add_to mat WHERE mat.msg_id = $1 AND lower(mat.addr) = lower($2))
@@ -1395,10 +1461,24 @@ func (h *MessageHandler) saveMessageData(fromAddr string, msgID int64, ext, data
 	if err := os.MkdirAll(dir, 0750); err != nil {
 		return "", fmt.Errorf("mkdir: %w", err)
 	}
-	filename := "data" + ext
-	path := filepath.Join(dir, filename)
-	if err := os.WriteFile(path, []byte(data), 0640); err != nil {
-		return "", fmt.Errorf("write: %w", err)
+	f, err := os.CreateTemp(dir, "data-*"+ext)
+	if err != nil {
+		return "", err
+	}
+	path := f.Name()
+	if err = f.Chmod(0640); err == nil {
+		_, err = f.WriteString(data)
+	}
+	if err == nil {
+		err = f.Sync()
+	}
+	closeErr := f.Close()
+	if err == nil {
+		err = closeErr
+	}
+	if err != nil {
+		_ = os.Remove(path)
+		return "", err
 	}
 	return path, nil
 }
@@ -1682,7 +1762,7 @@ func (h *MessageHandler) validateParent(c *gin.Context, pid *int64) bool {
 		return true
 	}
 	var terminal bool
-	err := h.DB.Pool.QueryRow(c.Request.Context(), "SELECT is_terminal FROM msg WHERE id = $1", *pid).Scan(&terminal)
+	err := h.q().QueryRow(c.Request.Context(), "SELECT is_terminal FROM msg WHERE id = $1", *pid).Scan(&terminal)
 	if errors.Is(err, pgx.ErrNoRows) {
 		c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("PID %d not found", *pid)})
 		return false
diff --git a/internal/handlers/reactions.go b/internal/handlers/reactions.go
index 525b617..c7b2b73 100644
--- a/internal/handlers/reactions.go
+++ b/internal/handlers/reactions.go
@@ -3,7 +3,9 @@ package handlers
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"errors"
+	"github.com/markmnl/fmsgd/pkg/message"
 	"io"
 	"log"
 	"net/http"
@@ -98,7 +100,7 @@ func (h *MessageHandler) reactionSubject(ctx context.Context, msgID int64) (int6
 	var noReply, terminal, important, hasAddTo bool
 	var mediaType, dataPath string
 	var size, attachments int
-	err := h.DB.Pool.QueryRow(ctx, `
+	err := h.q().QueryRow(ctx, `
 		SELECT m.pid, m.no_reply, m.is_terminal, m.is_important, m.type, m.size, m.filepath,
 		       (SELECT COUNT(*) FROM msg_attachment a WHERE a.msg_id = m.id),
 		       EXISTS (SELECT 1 FROM msg_add_to_batch b WHERE b.msg_id = m.id)
@@ -192,7 +194,7 @@ func (h *MessageHandler) loadReactionRows(ctx context.Context, subjectIDs []int6
 	if len(subjectIDs) == 0 {
 		return nil, nil
 	}
-	rows, err := h.DB.Pool.Query(ctx, `
+	rows, err := h.q().Query(ctx, `
 		SELECT r.pid, r.id, r.from_addr, r.time_sent, r.sha256, r.size, r.filepath
 		FROM msg r
 		WHERE r.pid = ANY($1)
@@ -293,7 +295,7 @@ type reactInput struct {
 // idempotent and sends nothing.
 func (h *MessageHandler) React(c *gin.Context) {
 	identity := middleware.GetIdentity(c)
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -360,10 +362,10 @@ func (h *MessageHandler) React(c *gin.Context) {
 	}
 	switch {
 	case current != nil && current.emoji == want:
-		c.JSON(http.StatusOK, gin.H{"id": current.msgID, "time": current.time})
+		c.JSON(http.StatusOK, gin.H{"id": current.msgID, "time": current.time, "sha256": hex.EncodeToString(current.hash)})
 		return
 	case current == nil && want == "":
-		c.JSON(http.StatusOK, gin.H{"id": nil, "time": nil})
+		c.JSON(http.StatusOK, gin.H{"id": nil, "time": nil, "sha256": nil})
 		return
 	}
 
@@ -382,8 +384,22 @@ func (h *MessageHandler) React(c *gin.Context) {
 		}
 	}
 
+	parentHash := subject.SHA256
+	if !sameAddr(subject.From, identity) && !isRecipient(subject.To, identity) {
+		parentHash = nil
+		for _, b := range subject.AddTo {
+			if isRecipient(b.To, identity) {
+				parentHash = b.SHA256
+				break
+			}
+		}
+	}
+	if parentHash == nil {
+		c.JSON(500, gin.H{"error": "reaction parent has no finalized identity"})
+		return
+	}
 	now := float64(time.Now().UnixMicro()) / 1e6
-	tx, err := h.DB.Pool.Begin(ctx)
+	tx, err := h.q().Begin(ctx)
 	if err != nil {
 		log.Printf("react to %d: begin: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to send reaction"})
@@ -393,10 +409,10 @@ func (h *MessageHandler) React(c *gin.Context) {
 
 	var reactionID int64
 	if err := tx.QueryRow(ctx,
-		`INSERT INTO msg (version, pid, no_reply, is_important, is_deflate, is_terminal, from_addr, topic, type, size, filepath, time_sent)
-		 VALUES (1, $1, true, false, false, true, $2, '', $3, $4, '', $5)
+		`INSERT INTO msg (version, pid, psha256, no_reply, is_important, is_deflate, is_terminal, from_addr, topic, type, size, filepath, time_sent)
+		 VALUES (1, $1, decode($5,'hex'), true, false, false, true, $2, '', $3, $4, '', NULL)
 		 RETURNING id`,
-		msgID, identity, reactionMediaType, len(want), now,
+		msgID, identity, reactionMediaType, len(want), parentHash,
 	).Scan(&reactionID); err != nil {
 		log.Printf("react to %d: insert: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to send reaction"})
@@ -408,6 +424,7 @@ func (h *MessageHandler) React(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to save reaction"})
 		return
 	}
+	createdFile(c, dataPath)
 	if _, err := tx.Exec(ctx, "UPDATE msg SET filepath = $1 WHERE id = $2", dataPath, reactionID); err != nil {
 		log.Printf("react to %d: update filepath: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to send reaction"})
@@ -420,6 +437,14 @@ func (h *MessageHandler) React(c *gin.Context) {
 			return
 		}
 	}
+	var files message.Files
+	onRollback(c, func() { files.Cleanup() })
+	hash, err := message.Finalize(ctx, finalizationTx{tx}, reactionID, now, &files)
+	if err != nil {
+		log.Printf("finalize reaction: %v", err)
+		c.JSON(500, gin.H{"error": "failed to finalize reaction"})
+		return
+	}
 	if err := tx.Commit(ctx); err != nil {
 		log.Printf("react to %d: commit: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to send reaction"})
@@ -428,7 +453,11 @@ func (h *MessageHandler) React(c *gin.Context) {
 
 	// fmsgd's outbound sender skips the local domain, so local recipients
 	// have their delivery resolved here, as Send does.
-	h.resolveLocalDelivery(ctx, "msg_to", reactionID, h.LocalDomain, recipients)
+	afterCommit(c, func() {
+		plain := *h
+		plain.query = nil
+		plain.resolveLocalDelivery(ctx, "msg_to", reactionID, h.LocalDomain, recipients)
+	})
 
-	c.JSON(http.StatusCreated, gin.H{"id": reactionID, "time": now})
+	c.JSON(http.StatusCreated, gin.H{"id": reactionID, "time": now, "sha256": hex.EncodeToString(hash)})
 }
diff --git a/internal/handlers/references.go b/internal/handlers/references.go
new file mode 100644
index 0000000..696683f
--- /dev/null
+++ b/internal/handlers/references.go
@@ -0,0 +1,122 @@
+package handlers
+
+import (
+	"bytes"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"github.com/jackc/pgx/v5"
+	"github.com/markmnl/fmsg-webapi/internal/middleware"
+)
+
+func decodeHash(s string) ([]byte, error) {
+	if len(s) != 64 {
+		return nil, errors.New("hash must contain 64 hexadecimal characters")
+	}
+	return hex.DecodeString(s)
+}
+
+func resolveID(c *gin.Context, q queries) (int64, bool) {
+	s := c.Param("id")
+	if len(s) != 64 {
+		return parseID(c)
+	}
+	hash, err := decodeHash(s)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid message sha256"})
+		return 0, false
+	}
+	var id int64
+	err = q.QueryRow(c.Request.Context(), `SELECT id FROM msg WHERE sha256=$1`, hash).Scan(&id)
+	if errors.Is(err, pgx.ErrNoRows) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "message not found"})
+		return 0, false
+	}
+	if err != nil {
+		c.JSON(500, gin.H{"error": "failed to resolve message"})
+		return 0, false
+	}
+	return id, true
+}
+
+// Resolve request pid separately from response pid, preserving the numeric
+// response contract and the exact parent hash when it identifies a batch.
+func (h *MessageHandler) resolveParent(c *gin.Context, in *messageInput) bool {
+	in.PID = nil
+	in.PSHA256 = nil
+	in.SHA256 = nil
+	raw := bytes.TrimSpace(in.Parent)
+	if len(raw) == 0 || bytes.Equal(raw, []byte("null")) {
+		return true
+	}
+	var id int64
+	var hash []byte
+	var err error
+	if raw[0] == '"' {
+		var value string
+		err = json.Unmarshal(raw, &value)
+		if err == nil {
+			hash, err = decodeHash(value)
+		}
+	} else {
+		id, err = strconv.ParseInt(string(raw), 10, 64)
+		if id <= 0 {
+			err = errors.New("parent id must be positive")
+		}
+	}
+	if err != nil {
+		c.JSON(400, gin.H{"error": "pid must be a positive integer or a 64-character sha256 string"})
+		return false
+	}
+	ctx := c.Request.Context()
+	var canonical []byte
+	var sent *float64
+	var terminal bool
+	var batchID *int64
+	if hash != nil {
+		err = h.q().QueryRow(ctx, `SELECT m.id,m.sha256,m.time_sent,m.is_terminal,b.id
+		 FROM msg m LEFT JOIN msg_add_to_batch b ON b.msg_id=m.id AND b.sha256=$1
+		 WHERE m.sha256=$1 OR b.id IS NOT NULL ORDER BY b.id NULLS FIRST LIMIT 1`, hash).Scan(&id, &canonical, &sent, &terminal, &batchID)
+	} else {
+		err = h.q().QueryRow(ctx, `SELECT sha256,time_sent,is_terminal FROM msg WHERE id=$1`, id).Scan(&canonical, &sent, &terminal)
+	}
+	if errors.Is(err, pgx.ErrNoRows) {
+		c.JSON(400, gin.H{"error": "parent message not found"})
+		return false
+	}
+	if err != nil {
+		c.JSON(500, gin.H{"error": "failed to resolve parent"})
+		return false
+	}
+	var participant bool
+	err = h.q().QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM msg WHERE id=$1 AND lower(from_addr)=lower($2))
+	 OR EXISTS(SELECT 1 FROM msg_to WHERE msg_id=$1 AND lower(addr)=lower($2))
+	 OR EXISTS(SELECT 1 FROM msg_add_to WHERE msg_id=$1 AND batch_id=$3 AND lower(addr)=lower($2))`, id, middleware.GetIdentity(c), batchID).Scan(&participant)
+	if err != nil {
+		c.JSON(500, gin.H{"error": "failed to validate parent participation"})
+		return false
+	}
+	if !participant {
+		c.JSON(403, gin.H{"error": "not a participant of this parent; added recipients must use their batch sha256"})
+		return false
+	}
+	if sent == nil || terminal {
+		c.JSON(409, gin.H{"error": "parent is a draft or terminal message"})
+		return false
+	}
+	if hash == nil {
+		hash = canonical
+	}
+	if len(hash) != 32 {
+		c.JSON(500, gin.H{"error": "parent has no finalized identity"})
+		return false
+	}
+	encoded := hex.EncodeToString(hash)
+	in.PID = &id
+	in.PSHA256 = &encoded
+	return true
+}
diff --git a/internal/handlers/thread.go b/internal/handlers/thread.go
index 55ecfff..3484585 100644
--- a/internal/handlers/thread.go
+++ b/internal/handlers/thread.go
@@ -45,6 +45,8 @@ type threadAttachment struct {
 }
 
 type threadMessage struct {
+	SHA256        *string             `json:"sha256,omitempty"`
+	PSHA256       *string             `json:"psha256,omitempty"`
 	ID            int64               `json:"id"`
 	Visible       bool                `json:"visible"`
 	Version       int                 `json:"version,omitempty"`
@@ -155,7 +157,7 @@ func loadThreadRelations(ctx context.Context, tx pgx.Tx, messages []threadMessag
 	}
 
 	rows, err = tx.Query(ctx, `
-		SELECT b.msg_id, b.id, b.add_to_from, b.time_added, a.addr
+		SELECT b.msg_id, b.id, b.add_to_from, b.time_added, encode(b.sha256,'hex'), a.addr
 		FROM msg_add_to_batch b
 		LEFT JOIN msg_add_to a ON a.batch_id = b.id
 		WHERE b.msg_id = ANY($1)
@@ -168,14 +170,15 @@ func loadThreadRelations(ctx context.Context, tx pgx.Tx, messages []threadMessag
 		var msgID, batchID int64
 		var from string
 		var added float64
+		var hash *string
 		var addr *string
-		if err = rows.Scan(&msgID, &batchID, &from, &added, &addr); err != nil {
+		if err = rows.Scan(&msgID, &batchID, &from, &added, &hash, &addr); err != nil {
 			rows.Close()
 			return err
 		}
 		idx, ok := batchIndexes[batchID]
 		if !ok {
-			byID[msgID].AddTo = append(byID[msgID].AddTo, models.AddToBatch{BatchID: batchID, AddToFrom: from, Time: added})
+			byID[msgID].AddTo = append(byID[msgID].AddTo, models.AddToBatch{BatchID: batchID, AddToFrom: from, Time: added, SHA256: hash})
 			idx = len(byID[msgID].AddTo) - 1
 			batchIndexes[batchID] = idx
 		}
@@ -269,13 +272,13 @@ func (h *MessageHandler) ThreadText(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"})
 		return
 	}
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
 	ctx := c.Request.Context()
 
-	rows, err := h.DB.Pool.Query(ctx, `
+	rows, err := h.q().Query(ctx, `
 		WITH RECURSIVE chain AS (
 			SELECT id, pid, from_addr, time_sent, type, size, filepath, 0 AS depth
 			FROM msg WHERE id = $1
@@ -363,7 +366,7 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"})
 		return
 	}
-	msgID, ok := parseID(c)
+	msgID, ok := resolveID(c, h.q())
 	if !ok {
 		return
 	}
@@ -379,18 +382,18 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) {
 	rows, err := tx.Query(ctx, `
 		WITH RECURSIVE chain AS (
 			SELECT id, version, pid, no_reply, is_important, is_deflate, is_terminal, time_sent,
-			       from_addr, topic, type, size, filepath, sha256, 0 AS depth
+			       from_addr, topic, type, size, filepath, sha256, psha256, 0 AS depth
 			FROM msg WHERE id = $1
 			UNION ALL
 			SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate,
 			       m.is_terminal, m.time_sent, m.from_addr, m.topic, m.type, m.size, m.filepath,
-			       m.sha256, c.depth + 1
+			       m.sha256, m.psha256, c.depth + 1
 			FROM msg m JOIN chain c ON m.id = c.pid
 			WHERE c.depth + 1 < $3
 		)
 		SELECT c.id, c.version, c.pid, c.no_reply, c.is_important, c.is_deflate,
 		       c.is_terminal, c.time_sent, c.from_addr, c.topic, c.type, c.size, c.filepath,
-		       encode(c.sha256, 'hex'),
+		       encode(c.sha256, 'hex'), encode(c.psha256, 'hex'),
 		       (lower(c.from_addr) = ANY($2)
 		        OR EXISTS (SELECT 1 FROM msg_to t WHERE t.msg_id = c.id AND lower(t.addr) = ANY($2))
 		        OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = c.id AND lower(a.addr) = ANY($2)))
@@ -406,13 +409,14 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) {
 		var hash *string
 		if err = rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important,
 			&m.Deflate, &m.Terminal, &m.Time, &m.From, &m.Topic, &m.Type, &m.Size, &m.dataPath,
-			&hash, &m.Visible); err != nil {
+			&hash, &m.PSHA256, &m.Visible); err != nil {
 			rows.Close()
 			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"})
 			return
 		}
 		if hash != nil {
 			m.MessageSHA256 = *hash
+			m.SHA256 = hash
 		}
 		messages = append(messages, m)
 	}
diff --git a/internal/handlers/transaction.go b/internal/handlers/transaction.go
new file mode 100644
index 0000000..ef5aed1
--- /dev/null
+++ b/internal/handlers/transaction.go
@@ -0,0 +1,151 @@
+package handlers
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"log"
+	"net/http"
+	"os"
+
+	"github.com/gin-gonic/gin"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/markmnl/fmsg-webapi/internal/db"
+	"github.com/markmnl/fmsgd/pkg/message"
+)
+
+type queries interface {
+	Query(context.Context, string, ...any) (pgx.Rows, error)
+	QueryRow(context.Context, string, ...any) pgx.Row
+	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
+	Begin(context.Context) (pgx.Tx, error)
+}
+
+func (h *MessageHandler) q() queries {
+	if h.query != nil {
+		return h.query
+	}
+	return h.DB.Pool
+}
+func (h *AttachmentHandler) q() queries {
+	if h.query != nil {
+		return h.query
+	}
+	return h.DB.Pool
+}
+
+type finalizationTx struct{ pgx.Tx }
+
+func (t finalizationTx) QueryRow(c context.Context, q string, a ...any) message.Row {
+	return t.Tx.QueryRow(c, q, a...)
+}
+func (t finalizationTx) Query(c context.Context, q string, a ...any) (message.Rows, error) {
+	return t.Tx.Query(c, q, a...)
+}
+func (t finalizationTx) Exec(c context.Context, q string, a ...any) error {
+	_, e := t.Tx.Exec(c, q, a...)
+	return e
+}
+
+// Atomic gives every message mutation one transaction and one message lock.
+// A request-local handler copy prevents sharing transactions across requests.
+func (h *MessageHandler) Atomic(next func(*MessageHandler, *gin.Context)) gin.HandlerFunc {
+	return func(c *gin.Context) { atomic(c, h.DB, func(tx pgx.Tx) { copy := *h; copy.query = tx; next(&copy, c) }) }
+}
+func (h *AttachmentHandler) Atomic(next func(*AttachmentHandler, *gin.Context)) gin.HandlerFunc {
+	return func(c *gin.Context) { atomic(c, h.DB, func(tx pgx.Tx) { copy := *h; copy.query = tx; next(&copy, c) }) }
+}
+
+type transactionFiles struct {
+	rollback []func()
+	commit   []func()
+}
+
+func onRollback(c *gin.Context, f func()) {
+	if v, ok := c.Get("messageTransactionFiles"); ok {
+		v.(*transactionFiles).rollback = append(v.(*transactionFiles).rollback, f)
+	}
+}
+func afterCommit(c *gin.Context, f func()) {
+	if v, ok := c.Get("messageTransactionFiles"); ok {
+		v.(*transactionFiles).commit = append(v.(*transactionFiles).commit, f)
+	} else {
+		f()
+	}
+}
+func createdFile(c *gin.Context, path string) { onRollback(c, func() { _ = os.Remove(path) }) }
+
+func atomic(c *gin.Context, database *db.DB, next func(pgx.Tx)) {
+	ctx := c.Request.Context()
+	tx, err := database.Pool.Begin(ctx)
+	if err != nil {
+		c.JSON(500, gin.H{"error": "failed to begin message transaction"})
+		return
+	}
+	defer tx.Rollback(ctx)
+	if c.Param("id") != "" {
+		id, ok := resolveID(c, tx)
+		if !ok {
+			return
+		}
+		var locked int64
+		if err = tx.QueryRow(ctx, `SELECT id FROM msg WHERE id=$1 FOR UPDATE`, id).Scan(&locked); err != nil && !errors.Is(err, pgx.ErrNoRows) {
+			c.JSON(500, gin.H{"error": "failed to lock message"})
+			return
+		}
+	}
+	files := &transactionFiles{}
+	c.Set("messageTransactionFiles", files)
+	committed := false
+	commitAttempted := false
+	defer func() {
+		if !committed && !commitAttempted {
+			for _, f := range files.rollback {
+				f()
+			}
+		}
+	}()
+	original := c.Writer
+	buffer := &mutationResponse{ResponseWriter: original, status: http.StatusOK}
+	c.Writer = buffer
+	defer func() { c.Writer = original }()
+	next(tx)
+	c.Writer = original
+	if buffer.status < 400 {
+		// A lost commit acknowledgement cannot prove rollback. Retain files
+		// rather than risk deleting payloads referenced by a committed row.
+		commitAttempted = true
+		if err = tx.Commit(ctx); err != nil {
+			log.Printf("commit message mutation: %v", err)
+			c.JSON(500, gin.H{"error": "failed to commit message mutation"})
+			return
+		}
+		committed = true
+		for _, f := range files.commit {
+			f()
+		}
+	} else {
+		_ = tx.Rollback(ctx)
+	}
+	original.WriteHeader(buffer.status)
+	_, _ = io.Copy(original, &buffer.body)
+}
+
+// HTTP success is buffered until commit; errors discard the transaction.
+// Only bounded JSON/no-content mutation responses use this writer.
+type mutationResponse struct {
+	gin.ResponseWriter
+	body   bytes.Buffer
+	status int
+}
+
+func (w *mutationResponse) WriteHeader(status int)            { w.status = status }
+func (w *mutationResponse) WriteHeaderNow()                   {}
+func (w *mutationResponse) Write(b []byte) (int, error)       { return w.body.Write(b) }
+func (w *mutationResponse) WriteString(s string) (int, error) { return w.body.WriteString(s) }
+func (w *mutationResponse) Status() int                       { return w.status }
+func (w *mutationResponse) Size() int                         { return w.body.Len() }
+func (w *mutationResponse) Written() bool                     { return w.body.Len() > 0 }
+func (w *mutationResponse) Flush()                            {}
diff --git a/internal/models/models.go b/internal/models/models.go
index c8f3118..d77ee13 100644
--- a/internal/models/models.go
+++ b/internal/models/models.go
@@ -18,6 +18,7 @@ type RecipientDelivery struct {
 // the recipients added in one POST /fmsg/:id/add-to call, who added them
 // (add_to_from), and when (time).
 type AddToBatch struct {
+	SHA256     *string             `json:"sha256"`
 	BatchID    int64               `json:"batch_id"`
 	AddToFrom  string              `json:"add_to_from"`
 	To         []string            `json:"to"`
@@ -27,6 +28,8 @@ type AddToBatch struct {
 
 // Message represents a fmsg message as exchanged over the HTTP API.
 type Message struct {
+	SHA256      *string             `json:"sha256"`
+	PSHA256     *string             `json:"psha256"`
 	Version     int                 `json:"version"`
 	HasPid      bool                `json:"has_pid"`
 	HasAddTo    bool                `json:"has_add_to"`