diff --git a/cmd/fmsg-backfill/main.go b/cmd/fmsg-backfill/main.go
deleted file mode 100644
index ab2e0bc..0000000
--- a/cmd/fmsg-backfill/main.go
+++ /dev/null
@@ -1,118 +0,0 @@
-// 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
-}
diff --git a/cmd/fmsg-backfill/migrate.go b/cmd/fmsg-backfill/migrate.go
deleted file mode 100644
index a3de0e8..0000000
--- a/cmd/fmsg-backfill/migrate.go
+++ /dev/null
@@ -1,493 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"database/sql"
-	"fmt"
-	"os"
-	"path/filepath"
-	"strings"
-
-	"github.com/markmnl/fmsgd/pkg/fmsg"
-)
-
-type migration struct {
-	tx       *sql.Tx
-	domain   string
-	files    []string
-	visiting map[int64]bool
-	done     map[int64]bool
-}
-
-type oldMessage struct {
-	id       int64
-	pid      *int64
-	time     *float64
-	h        *fmsg.Header // expanded payload paths from the old store
-	hash     []byte
-	wire     []byte
-	prepared []byte
-}
-
-type oldBatch struct {
-	id       int64
-	from     fmsg.Address
-	time     float64
-	to       []fmsg.Address
-	hash     []byte
-	prepared []byte
-}
-
-func address(s string) (fmsg.Address, error) {
-	p := strings.SplitN(s, "@", 3)
-	if len(p) != 3 || p[0] != "" || p[1] == "" || p[2] == "" || len(s) > 255 {
-		return fmsg.Address{}, fmt.Errorf("invalid stored address %q", s)
-	}
-	return fmsg.Address{User: p[1], Domain: p[2]}, nil
-}
-
-func (m *migration) ids(query string, args ...any) ([]int64, error) {
-	rows, err := m.tx.Query(query, args...)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var ids []int64
-	for rows.Next() {
-		var id int64
-		if err = rows.Scan(&id); err != nil {
-			return nil, err
-		}
-		ids = append(ids, id)
-	}
-	return ids, rows.Err()
-}
-
-func (m *migration) addresses(query string, id int64) ([]fmsg.Address, error) {
-	rows, err := m.tx.Query(query, id)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var list []fmsg.Address
-	for rows.Next() {
-		var raw string
-		if err = rows.Scan(&raw); err != nil {
-			return nil, err
-		}
-		a, err := address(raw)
-		if err != nil {
-			return nil, err
-		}
-		list = append(list, a)
-	}
-	return list, rows.Err()
-}
-
-func (m *migration) load(id int64) (*oldMessage, error) {
-	s := &oldMessage{id: id, h: &fmsg.Header{}}
-	var from string
-	var noReply, important, terminal bool
-	err := m.tx.QueryRow(`SELECT version,pid,psha256,no_reply,is_important,is_terminal,time_sent,from_addr,topic,type,size,filepath,sha256,wire_header,wire_message FROM msg WHERE id=$1`, id).Scan(&s.h.Version, &s.pid, &s.h.Pid, &noReply, &important, &terminal, &s.time, &from, &s.h.Topic, &s.h.Type, &s.h.Size, &s.h.Filepath, &s.hash, &s.wire, &s.prepared)
-	if err != nil {
-		return nil, err
-	}
-	s.h.From, err = address(from)
-	if err != nil {
-		return nil, err
-	}
-	if noReply {
-		s.h.Flags |= fmsg.FlagNoReply
-	}
-	if important {
-		s.h.Flags |= fmsg.FlagImportant
-	}
-	if terminal {
-		s.h.Flags |= fmsg.FlagTerminal
-	}
-	if s.time != nil {
-		s.h.Timestamp = *s.time
-	}
-	if s.pid != nil || len(s.h.Pid) > 0 {
-		s.h.Flags |= fmsg.FlagHasPid
-	}
-	s.h.To, err = m.addresses(`SELECT addr FROM msg_to WHERE msg_id=$1 ORDER BY id`, id)
-	if err != nil {
-		return nil, err
-	}
-	rows, err := m.tx.Query(`SELECT type,filename,filesize,filepath FROM msg_attachment WHERE msg_id=$1 ORDER BY position,filename`, id)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var a fmsg.AttachmentHeader
-		if err = rows.Scan(&a.Type, &a.Filename, &a.Size, &a.Filepath); err != nil {
-			return nil, err
-		}
-		s.h.Attachments = append(s.h.Attachments, a)
-	}
-	return s, rows.Err()
-}
-
-func (m *migration) batches(id int64) ([]oldBatch, error) {
-	ids, err := m.ids(`SELECT id FROM msg_add_to_batch WHERE msg_id=$1 ORDER BY id`, id)
-	if err != nil {
-		return nil, err
-	}
-	var batches []oldBatch
-	for _, id := range ids {
-		b := oldBatch{id: id}
-		var from string
-		if err = m.tx.QueryRow(`SELECT add_to_from,time_added,sha256,wire_message FROM msg_add_to_batch WHERE id=$1`, id).Scan(&from, &b.time, &b.hash, &b.prepared); err != nil {
-			return nil, err
-		}
-		b.from, err = address(from)
-		if err != nil {
-			return nil, err
-		}
-		b.to, err = m.addresses(`SELECT addr FROM msg_add_to WHERE batch_id=$1 ORDER BY id`, id)
-		if err != nil {
-			return nil, err
-		}
-		batches = append(batches, b)
-	}
-	return batches, nil
-}
-
-func (m *migration) message(id int64) error {
-	if m.done[id] {
-		return nil
-	}
-	if m.visiting[id] {
-		return fmt.Errorf("cyclic parent links at message %d", id)
-	}
-	m.visiting[id] = true
-	defer delete(m.visiting, id)
-	s, err := m.load(id)
-	if err != nil {
-		return err
-	}
-	if s.time == nil {
-		return fmt.Errorf("sent child references draft parent %d", id)
-	}
-	if s.pid != nil {
-		if err = m.message(*s.pid); err != nil {
-			return fmt.Errorf("parent %d: %w", *s.pid, err)
-		}
-		if len(s.h.Pid) == 0 {
-			if err = m.tx.QueryRow(`SELECT sha256 FROM msg WHERE id=$1`, *s.pid).Scan(&s.h.Pid); err != nil {
-				return err
-			}
-		}
-	}
-	// Old receivers sometimes kept wire sizes beside already expanded files.
-	// For published identities the full hash remains the authority: use actual
-	// expanded lengths when reconstructing, and only persist them after verification.
-	if len(s.hash) == 32 && len(s.prepared) == 0 {
-		if err = expandedSizes(s.h); err != nil {
-			return err
-		}
-	}
-	batches, err := m.batches(id)
-	if err != nil {
-		return err
-	}
-	var base, received *fmsg.Header
-	if len(s.prepared) > 0 {
-		base, err = fmsg.UnmarshalPrepared(s.prepared, s.hash)
-	} else if len(s.wire) > 0 {
-		received, err = decodeHeader(s.wire)
-		if err == nil && received.Flags&fmsg.FlagHasAddTo != 0 {
-			for _, b := range batches {
-				if len(b.prepared) > 0 {
-					candidate, e := fmsg.UnmarshalPrepared(b.prepared, b.hash)
-					if e != nil {
-						return e
-					}
-					if bytes.Equal(candidate.Encode(), s.wire) {
-						received = candidate
-						break
-					}
-				}
-			}
-		}
-		if err == nil && received.Filepath == "" {
-			received, err = m.restore(received, s.h)
-		}
-		if err == nil && received.Flags&fmsg.FlagHasAddTo == 0 {
-			base = received
-		} else if err == nil {
-			// The canonical row represents an original known only by the pid
-			// of the received add-to. Its original header is not recoverable.
-			if !bytes.Equal(received.Pid, s.hash) {
-				return fmt.Errorf("received batch pid differs from canonical identity")
-			}
-		}
-	} else {
-		// Rerunning against a fully migrated add-to-only original is valid.
-		for _, b := range batches {
-			if len(b.prepared) > 0 {
-				received, err = fmsg.UnmarshalPrepared(b.prepared, b.hash)
-				break
-			}
-		}
-		if received == nil && err == nil {
-			if len(s.hash) == 0 && !strings.EqualFold(s.h.From.Domain, m.domain) {
-				return fmt.Errorf("remote message has no published hash or wire header")
-			}
-			base, err = m.reconstruct(s.h, s.hash, batches)
-		}
-	}
-	if err != nil {
-		return err
-	}
-	if base != nil {
-		hash, err := base.GetMessageHash()
-		if err != nil {
-			return err
-		}
-		if len(s.hash) > 0 && !bytes.Equal(hash, s.hash) {
-			return fmt.Errorf("cannot reproduce published message hash; identity preserved")
-		}
-		s.hash = hash
-		data, err := fmsg.MarshalPrepared(base)
-		if err != nil {
-			return err
-		}
-		if _, err = m.tx.Exec(`UPDATE msg SET sha256=$2,psha256=$3,wire_header=$4,wire_message=$5,is_deflate=$6 WHERE id=$1`, id, hash, bytesOrNull(base.Pid), base.Encode(), string(data), base.Flags&fmsg.FlagDeflate != 0); err != nil {
-			return err
-		}
-	} else {
-		base = received
-	}
-	if base == nil || len(s.hash) != 32 {
-		return fmt.Errorf("missing canonical identity or payload representation")
-	}
-	// Early receivers stamped the arrival time on the batch row. An exact
-	// retained wire header supplies the sending timestamp. Only repair an
-	// unambiguous, unhashed batch; existing batch identities remain authoritative.
-	if received != nil && received.Flags&fmsg.FlagHasAddTo != 0 {
-		candidate := -1
-		matches := 0
-		exact := false
-		for i, b := range batches {
-			h := batchHeader(base, s.hash, b)
-			if bytes.Equal(h.Encode(), received.Encode()) {
-				exact = true
-				break
-			}
-			h.Timestamp = received.Timestamp
-			if len(b.hash) == 0 && bytes.Equal(h.Encode(), received.Encode()) {
-				candidate = i
-				matches++
-			}
-		}
-		if !exact && matches == 1 {
-			batches[candidate].time = received.Timestamp
-			if _, err = m.tx.Exec(`UPDATE msg_add_to_batch SET time_added=$2 WHERE id=$1`, batches[candidate].id, received.Timestamp); err != nil {
-				return err
-			}
-		}
-	}
-	matchedReceived := received == nil || received.Flags&fmsg.FlagHasAddTo == 0
-	for _, b := range batches {
-		h := batchHeader(base, s.hash, b)
-		if received != nil && bytes.Equal(h.Encode(), received.Encode()) {
-			matchedReceived = true
-		}
-		if len(b.prepared) > 0 {
-			h, err = fmsg.UnmarshalPrepared(b.prepared, b.hash)
-		} else {
-			h, err = selectTypes(h, b.hash)
-		}
-		if err != nil {
-			return fmt.Errorf("batch %d: %w", b.id, err)
-		}
-		if !bytes.Equal(h.Pid, s.hash) {
-			return fmt.Errorf("batch %d references a different original", b.id)
-		}
-		hash, err := h.GetMessageHash()
-		if err != nil {
-			return err
-		}
-		data, err := fmsg.MarshalPrepared(h)
-		if err != nil {
-			return err
-		}
-		if _, err = m.tx.Exec(`UPDATE msg_add_to_batch SET sha256=$2,wire_message=$3 WHERE id=$1`, b.id, hash, string(data)); err != nil {
-			return err
-		}
-	}
-	if !matchedReceived {
-		return fmt.Errorf("received wire header has no matching add-to batch")
-	}
-	if _, err = m.tx.Exec(`UPDATE msg SET size=$2 WHERE id=$1`, id, s.h.Size); err != nil {
-		return err
-	}
-	for _, a := range s.h.Attachments {
-		if _, err = m.tx.Exec(`UPDATE msg_attachment SET filesize=$3 WHERE msg_id=$1 AND filename=$2`, id, a.Filename, a.Size); err != nil {
-			return err
-		}
-	}
-	m.done[id] = true
-	return nil
-}
-
-func batchHeader(base *fmsg.Header, hash []byte, b oldBatch) *fmsg.Header {
-	h := base.Clone()
-	h.Flags |= fmsg.FlagHasPid | fmsg.FlagHasAddTo
-	h.Pid, h.Timestamp, h.Topic = hash, b.time, ""
-	h.AddToFrom, h.AddTo = &b.from, b.to
-	return h
-}
-
-// Local sends previously selected compression and common types during the
-// first network delivery. Try those historical forms only in this tool, and
-// accept a candidate only if it reproduces the entire existing message hash.
-func (m *migration) reconstruct(raw *fmsg.Header, expected []byte, batches []oldBatch) (*fmsg.Header, error) {
-	input := raw.Clone()
-	// Early local notes could have an empty recipient list. Preserve their
-	// recorded bytes; normal send validation still requires recipients.
-	if len(input.To) == 0 {
-		input.To = []fmsg.Address{input.From}
-	}
-	h, dir, err := fmsg.Prepare(input)
-	if err != nil {
-		return nil, err
-	}
-	h = h.Clone()
-	h.To = raw.To
-	if chosen, err := selectHistorical(h, expected, batches); err == nil {
-		m.files = append(m.files, dir)
-		return chosen, nil
-	}
-	_ = os.RemoveAll(dir)
-	h, dir, err = fmsg.Preserve(raw, filepath.Dir(raw.Filepath))
-	if err != nil {
-		return nil, err
-	}
-	chosen, err := selectHistorical(h, expected, batches)
-	if err != nil {
-		_ = os.RemoveAll(dir)
-		return nil, err
-	}
-	m.files = append(m.files, dir)
-	return chosen, nil
-}
-
-// An early sender could cache a canonical hash after adding batch fields but
-// before setting the pid flag. Only retain that form if recorded batch fields
-// reproduce the already published hash; never create this encoding for new IDs.
-func selectHistorical(h *fmsg.Header, expected []byte, batches []oldBatch) (*fmsg.Header, error) {
-	chosen, err := selectOriginal(h, expected)
-	if err == nil || len(expected) != 32 {
-		return chosen, err
-	}
-	for _, b := range batches {
-		candidate := h.Clone()
-		candidate.Flags = (candidate.Flags | fmsg.FlagHasAddTo) &^ fmsg.FlagHasPid
-		candidate.AddToFrom, candidate.AddTo = &b.from, b.to
-		if chosen, e := selectTypes(candidate, expected); e == nil {
-			return chosen, nil
-		}
-	}
-	return nil, err
-}
-
-// Some early local writers hashed a root-form header despite retaining a
-// relational parent link. Preserve that established identity and local link;
-// never select this form for a message without an existing hash to verify.
-func selectOriginal(h *fmsg.Header, expected []byte) (*fmsg.Header, error) {
-	chosen, err := selectTypes(h, expected)
-	if err == nil {
-		return chosen, nil
-	}
-	if len(expected) == 32 && h.Flags&fmsg.FlagHasPid != 0 && h.Flags&fmsg.FlagHasAddTo == 0 {
-		root := h.Clone()
-		root.Flags &^= fmsg.FlagHasPid
-		return selectTypes(root, expected)
-	}
-	return nil, err
-}
-
-func selectTypes(h *fmsg.Header, expected []byte) (*fmsg.Header, error) {
-	for _, mode := range []int{0, 1, 2} {
-		c := h.Clone()
-		switch mode {
-		case 1:
-			fmsg.ApplyCommonTypes(c)
-		case 2:
-			c.Flags &^= fmsg.FlagCommonType
-			for i := range c.Attachments {
-				c.Attachments[i].Flags &^= 1
-			}
-		}
-		hash, err := c.GetMessageHash()
-		if err != nil {
-			return nil, err
-		}
-		if len(expected) == 0 || bytes.Equal(expected, hash) {
-			return c, nil
-		}
-	}
-	return nil, fmt.Errorf("cannot reproduce published hash; identity preserved")
-}
-
-func (m *migration) validate() error {
-	var invalid int
-	err := m.tx.QueryRow(`SELECT count(*) FROM msg m WHERE
-		(m.time_sent IS NOT NULL AND (m.sha256 IS NULL OR octet_length(m.sha256)<>32 OR
-		 (m.wire_message IS NULL AND NOT EXISTS (SELECT 1 FROM msg_add_to_batch b WHERE b.msg_id=m.id AND b.wire_message IS NOT NULL)))) OR
-		(m.pid IS NOT NULL AND NOT EXISTS (SELECT 1 FROM msg p WHERE p.id=m.pid AND p.time_sent IS NOT NULL AND NOT p.is_terminal AND
-		 (m.psha256=p.sha256 OR EXISTS (SELECT 1 FROM msg_add_to_batch b WHERE b.msg_id=p.id AND b.sha256=m.psha256)))) OR
-		(m.time_sent IS NULL AND (m.sha256 IS NOT NULL OR m.wire_message IS NOT NULL))`).Scan(&invalid)
-	if err != nil {
-		return err
-	}
-	if invalid != 0 {
-		return fmt.Errorf("%d messages violate finalized identity or parent invariants", invalid)
-	}
-	err = m.tx.QueryRow(`SELECT count(*) FROM msg_add_to_batch b JOIN msg m ON m.id=b.msg_id WHERE
-		m.is_terminal OR (m.time_sent IS NOT NULL AND
-		(b.sha256 IS NULL OR octet_length(b.sha256)<>32 OR b.wire_message IS NULL))`).Scan(&invalid)
-	if err != nil {
-		return err
-	}
-	if invalid != 0 {
-		return fmt.Errorf("%d batches violate finalized identity invariants", invalid)
-	}
-	return nil
-}
-
-func bytesOrNull(b []byte) any {
-	if len(b) == 0 {
-		return nil
-	}
-	return b
-}
-
-func expandedSizes(h *fmsg.Header) error {
-	size := func(path string) (uint32, error) {
-		info, err := os.Stat(path)
-		if err != nil {
-			return 0, err
-		}
-		if !info.Mode().IsRegular() || info.Size() < 0 || info.Size() > int64(^uint32(0)) {
-			return 0, fmt.Errorf("invalid payload size: %s", path)
-		}
-		return uint32(info.Size()), nil
-	}
-	var err error
-	h.Size, err = size(h.Filepath)
-	if err != nil {
-		return err
-	}
-	for i := range h.Attachments {
-		h.Attachments[i].Size, err = size(h.Attachments[i].Filepath)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
diff --git a/cmd/fmsg-backfill/migrate_test.go b/cmd/fmsg-backfill/migrate_test.go
deleted file mode 100644
index cd36879..0000000
--- a/cmd/fmsg-backfill/migrate_test.go
+++ /dev/null
@@ -1,492 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"context"
-	"database/sql"
-	"fmt"
-	"io"
-	"net/url"
-	"os"
-	"path/filepath"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/markmnl/fmsgd/pkg/fmsg"
-)
-
-func previousStore(t *testing.T) *sql.DB {
-	t.Helper()
-	dsn := os.Getenv("FMSG_TEST_DATABASE_URL")
-	if dsn == "" {
-		t.Skip("set FMSG_TEST_DATABASE_URL to test the standalone migration")
-	}
-	admin, err := sql.Open("postgres", dsn)
-	if err != nil {
-		t.Fatal(err)
-	}
-	schema := fmt.Sprintf("backfill_%d", time.Now().UnixNano())
-	if _, err = admin.Exec("CREATE SCHEMA " + schema); err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(func() { admin.Exec("DROP SCHEMA " + schema + " CASCADE"); admin.Close() })
-	u, err := url.Parse(dsn)
-	if err != nil {
-		t.Fatal(err)
-	}
-	q := u.Query()
-	q.Set("search_path", schema)
-	u.RawQuery = q.Encode()
-	db, err := sql.Open("postgres", u.String())
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(func() { db.Close() })
-	dd, err := os.ReadFile("testdata/previous.sql")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if _, err = db.Exec(string(dd)); err != nil {
-		t.Fatal(err)
-	}
-	return db
-}
-
-func rawMessage(t *testing.T, domain, content string) *fmsg.Header {
-	t.Helper()
-	path := filepath.Join(t.TempDir(), "body")
-	if err := os.WriteFile(path, []byte(content), 0600); err != nil {
-		t.Fatal(err)
-	}
-	return &fmsg.Header{Version: 1, From: fmsg.Address{User: "alice", Domain: domain},
-		To: []fmsg.Address{{User: "bob", Domain: "example.com"}}, Timestamp: 1234.5,
-		Topic: "stored", Type: "text/plain;charset=UTF-8", Filepath: path, Size: uint32(len(content))}
-}
-
-func putOld(t *testing.T, db *sql.DB, raw *fmsg.Header, pid any, hash, header []byte) int64 {
-	t.Helper()
-	var id int64
-	if err := db.QueryRow(`INSERT INTO msg(version,pid,psha256,time_sent,from_addr,topic,type,size,filepath,sha256,wire_header,no_reply,is_important,is_terminal)
-		VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) RETURNING id`, raw.Version, pid, bytesOrNull(raw.Pid), raw.Timestamp, raw.From.ToString(), raw.Topic, raw.Type, raw.Size, raw.Filepath, bytesOrNull(hash), bytesOrNull(header),
-		raw.Flags&fmsg.FlagNoReply != 0, raw.Flags&fmsg.FlagImportant != 0, raw.Flags&fmsg.FlagTerminal != 0).Scan(&id); err != nil {
-		t.Fatal(err)
-	}
-	for _, a := range raw.To {
-		if _, err := db.Exec(`INSERT INTO msg_to(msg_id,addr,time_delivered,response_code) VALUES($1,$2,1235,200)`, id, a.ToString()); err != nil {
-			t.Fatal(err)
-		}
-	}
-	for i, a := range raw.Attachments {
-		if _, err := db.Exec(`INSERT INTO msg_attachment(msg_id,position,type,filename,filesize,filepath) VALUES($1,$2,$3,$4,$5,$6)`, id, i, a.Type, a.Filename, a.Size, a.Filepath); err != nil {
-			t.Fatal(err)
-		}
-	}
-	return id
-}
-
-func putBatch(t *testing.T, db *sql.DB, id int64, b oldBatch) int64 {
-	t.Helper()
-	var bid int64
-	if err := db.QueryRow(`INSERT INTO msg_add_to_batch(msg_id,add_to_from,time_added,sha256) VALUES($1,$2,$3,$4) RETURNING id`, id, b.from.ToString(), b.time, bytesOrNull(b.hash)).Scan(&bid); err != nil {
-		t.Fatal(err)
-	}
-	for _, a := range b.to {
-		if _, err := db.Exec(`INSERT INTO msg_add_to(msg_id,batch_id,addr) VALUES($1,$2,$3)`, id, bid, a.ToString()); err != nil {
-			t.Fatal(err)
-		}
-	}
-	return bid
-}
-
-func prepared(t *testing.T, raw *fmsg.Header) *fmsg.Header {
-	t.Helper()
-	h, dir, err := fmsg.Prepare(raw)
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(func() { os.RemoveAll(dir) })
-	return h
-}
-func hashOf(t *testing.T, h *fmsg.Header) []byte {
-	t.Helper()
-	hash, err := h.GetMessageHash()
-	if err != nil {
-		t.Fatal(err)
-	}
-	return hash
-}
-
-func TestStandaloneMigration(t *testing.T) {
-	db := previousStore(t)
-	rootRaw := rawMessage(t, "example.com", "local root")
-	root := putOld(t, db, rootRaw, nil, nil, nil)
-	childRaw := rawMessage(t, "example.com", "local reply")
-	childRaw.Flags = fmsg.FlagHasPid
-	child := putOld(t, db, childRaw, root, nil, nil)
-	localBatch := putBatch(t, db, root, oldBatch{from: rootRaw.From, time: 1300, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}})
-
-	// Published string-form hash must remain string-form, even though new
-	// finalization chooses common type IDs.
-	oldString := rawMessage(t, "example.com", "published")
-	oldHash := hashOf(t, oldString)
-	published := putOld(t, db, oldString, nil, oldHash, nil)
-
-	localCompressed := rawMessage(t, "example.com", strings.Repeat("published compressed ", 300))
-	localWire := prepared(t, localCompressed)
-	localHash := hashOf(t, localWire)
-	compressed := putOld(t, db, localCompressed, nil, localHash, nil)
-	publishedBatch := oldBatch{from: oldString.From, time: 1301, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}}
-	publishedBatch.hash = hashOf(t, batchHeader(oldString, oldHash, publishedBatch))
-	publishedBatchID := putBatch(t, db, published, publishedBatch)
-
-	// Pending drafts and draft batches remain editable; an existing draft
-	// reply gains its newly finalized parent's protocol identity.
-	var draft int64
-	if err := db.QueryRow(`INSERT INTO msg(version,pid,from_addr,topic,type,size,filepath) VALUES(1,$1,'@alice@example.com','','text/plain',0,'') RETURNING id`, root).Scan(&draft); err != nil {
-		t.Fatal(err)
-	}
-	draftBatch := putBatch(t, db, draft, oldBatch{from: rootRaw.From, time: 1302, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}})
-
-	// Received compression and mixed type encodings must preserve the exact
-	// wire header. Expanded body and attachment files are all the old store has.
-	remote := rawMessage(t, "example.org", strings.Repeat("compressible content ", 400))
-	att := rawMessage(t, "example.org", strings.Repeat("attachment ", 300))
-	remote.Attachments = []fmsg.AttachmentHeader{{Type: "text/plain;charset=UTF-8", Filename: "note.txt", Size: att.Size, Filepath: att.Filepath}}
-	remoteWire := prepared(t, remote).Clone()
-	remoteWire.Attachments[0].Flags &^= 1
-	remoteHash := hashOf(t, remoteWire)
-	received := putOld(t, db, remote, nil, remoteHash, remoteWire.Encode())
-
-	// First delivery through add-to retains the canonical hash carried as pid,
-	// and prepares the batch without inventing an original header.
-	batchRaw := rawMessage(t, "example.org", strings.Repeat("forwarded content ", 400))
-	batchBase := prepared(t, batchRaw)
-	canonicalHash := hashOf(t, batchBase)
-	b := oldBatch{from: batchRaw.From, time: 1400, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}}
-	batchWire := batchHeader(batchBase, canonicalHash, b)
-	b.hash = hashOf(t, batchWire)
-	batchRaw.Timestamp = b.time
-	addToOnly := putOld(t, db, batchRaw, nil, canonicalHash, batchWire.Encode())
-	receivedBatch := putBatch(t, db, addToOnly, b)
-
-	// A dry run exercises the full conversion, then removes its columns and files.
-	if err := migrate(context.Background(), db, "example.com", false, io.Discard); err != nil {
-		t.Fatal(err)
-	}
-	var count int
-	if err := db.QueryRow(`SELECT count(*) FROM information_schema.columns WHERE table_schema=current_schema() AND column_name='wire_message'`).Scan(&count); err != nil || count != 0 {
-		t.Fatalf("dry run changed schema: %d %v", count, err)
-	}
-	for _, raw := range []*fmsg.Header{rootRaw, childRaw, oldString} {
-		matches, _ := filepath.Glob(filepath.Join(filepath.Dir(raw.Filepath), ".fmsg-wire-*"))
-		if len(matches) != 0 {
-			t.Fatal("dry run left files", matches)
-		}
-	}
-	if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-		t.Fatal(err)
-	}
-	var rootHash []byte
-	var snapshots = make(map[int64][]byte)
-	for _, id := range []int64{root, child, published, compressed, received, addToOnly} {
-		var hash, data, parent []byte
-		var stamp float64
-		if err := db.QueryRow(`SELECT sha256,wire_message,psha256,time_sent FROM msg WHERE id=$1`, id).Scan(&hash, &data, &parent, &stamp); err != nil {
-			t.Fatal(err)
-		}
-		if id == root {
-			rootHash = hash
-		}
-		if id == child && !bytes.Equal(parent, rootHash) {
-			t.Fatal("reply parent not backfilled")
-		}
-		if id == published && !bytes.Equal(hash, oldHash) {
-			t.Fatal("published string hash changed")
-		}
-		if id == compressed && !bytes.Equal(hash, localHash) {
-			t.Fatal("published compressed hash changed")
-		}
-		if id == received && !bytes.Equal(hash, remoteHash) {
-			t.Fatal("received hash changed")
-		}
-		if id == addToOnly {
-			if !bytes.Equal(hash, canonicalHash) || len(data) != 0 || stamp != 1400 {
-				t.Fatal("invented original identity")
-			}
-		} else {
-			if stamp != 1234.5 {
-				t.Fatal("timestamp changed")
-			}
-			if _, err := fmsg.UnmarshalPrepared(data, hash); err != nil {
-				t.Fatal(err)
-			}
-		}
-		snapshots[id] = data
-	}
-	for _, id := range []int64{localBatch, publishedBatchID, receivedBatch} {
-		var hash, data []byte
-		if err := db.QueryRow(`SELECT sha256,wire_message FROM msg_add_to_batch WHERE id=$1`, id).Scan(&hash, &data); err != nil {
-			t.Fatal(err)
-		}
-		if _, err := fmsg.UnmarshalPrepared(data, hash); err != nil {
-			t.Fatal(err)
-		}
-		if id == publishedBatchID && !bytes.Equal(hash, publishedBatch.hash) {
-			t.Fatal("published batch hash changed")
-		}
-		if id == receivedBatch && !bytes.Equal(hash, b.hash) {
-			t.Fatal("received batch identity changed")
-		}
-	}
-	var draftHash, draftParent, draftSnapshot []byte
-	var draftTime *float64
-	if err := db.QueryRow(`SELECT sha256,psha256,wire_message,time_sent FROM msg WHERE id=$1`, draft).Scan(&draftHash, &draftParent, &draftSnapshot, &draftTime); err != nil {
-		t.Fatal(err)
-	}
-	if len(draftHash) != 0 || len(draftSnapshot) != 0 || draftTime != nil || !bytes.Equal(draftParent, rootHash) {
-		t.Fatal("draft identity/state changed")
-	}
-	if err := db.QueryRow(`SELECT sha256,wire_message FROM msg_add_to_batch WHERE id=$1`, draftBatch).Scan(&draftHash, &draftSnapshot); err != nil {
-		t.Fatal(err)
-	}
-	if len(draftHash) != 0 || len(draftSnapshot) != 0 {
-		t.Fatal("prematurely finalized draft batch")
-	}
-
-	if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-		t.Fatal("rerun", err)
-	}
-	for id, expected := range snapshots {
-		var got []byte
-		db.QueryRow(`SELECT wire_message FROM msg WHERE id=$1`, id).Scan(&got)
-		if !bytes.Equal(expected, got) {
-			t.Fatal("rerun changed snapshot", id)
-		}
-	}
-	if _, err := db.Exec(`UPDATE msg SET topic='changed' WHERE id=$1`, root); err == nil {
-		t.Fatal("migration did not install strict immutability")
-	}
-	if _, err := db.Exec(`UPDATE msg SET wire_message=NULL WHERE id=$1`, root); err == nil {
-		t.Fatal("migration permits clearing snapshot")
-	}
-	if _, err := db.Exec(`UPDATE msg_to SET time_read=2000 WHERE msg_id=$1`, root); err != nil {
-		t.Fatal("receipt blocked", err)
-	}
-	if _, err := db.Exec(`INSERT INTO msg(version,time_sent,from_addr,topic,type,size,filepath) VALUES(1,1234,'@alice@example.com','','text/plain',0,'')`); err == nil {
-		t.Fatal("migration permits sent message without identity")
-	}
-}
-
-func TestMigrationFailureRollsBackEverything(t *testing.T) {
-	for _, cause := range []string{"missing file", "hash mismatch", "hashed child without parent hash"} {
-		t.Run(cause, func(t *testing.T) {
-			db := previousStore(t)
-			raw := rawMessage(t, "example.com", "good")
-			root := putOld(t, db, raw, nil, nil, nil)
-			bad := rawMessage(t, "example.com", "bad")
-			switch cause {
-			case "missing file":
-				putOld(t, db, bad, nil, nil, nil)
-				os.Remove(bad.Filepath)
-			case "hash mismatch":
-				putOld(t, db, bad, nil, bytes.Repeat([]byte{7}, 32), nil)
-			case "hashed child without parent hash":
-				putOld(t, db, bad, root, bytes.Repeat([]byte{7}, 32), nil)
-			}
-			if err := migrate(context.Background(), db, "example.com", true, io.Discard); err == nil {
-				t.Fatal("unsafe migration succeeded")
-			}
-			var hash []byte
-			if err := db.QueryRow(`SELECT sha256 FROM msg WHERE id=$1`, root).Scan(&hash); err != nil || hash != nil {
-				t.Fatal("partial data conversion", err)
-			}
-			var count int
-			db.QueryRow(`SELECT count(*) FROM information_schema.columns WHERE table_schema=current_schema() AND column_name='wire_message'`).Scan(&count)
-			if count != 0 {
-				t.Fatal("partial schema conversion")
-			}
-			files, _ := filepath.Glob(filepath.Join(filepath.Dir(raw.Filepath), ".fmsg-wire-*"))
-			if len(files) > 0 {
-				t.Fatal("rollback left staged files", files)
-			}
-		})
-	}
-}
-
-func TestDecodeHeaderRejectsTruncation(t *testing.T) {
-	h := prepared(t, rawMessage(t, "example.com", strings.Repeat("content ", 200)))
-	data := h.Encode()
-	for i := range data {
-		if _, err := decodeHeader(data[:i]); err == nil {
-			t.Fatalf("accepted prefix %d", i)
-		}
-	}
-	if _, err := decodeHeader(append(data, 0)); err == nil {
-		t.Fatal("accepted trailing data")
-	}
-}
-
-func TestMigrationExpandedFilesWithOldWireSizes(t *testing.T) {
-	for _, retainHeader := range []bool{false, true} {
-		t.Run(fmt.Sprintf("header=%v", retainHeader), func(t *testing.T) {
-			db := previousStore(t)
-			raw := rawMessage(t, "example.org", strings.Repeat("body compression ", 200))
-			att := rawMessage(t, "example.org", strings.Repeat("attachment compression ", 200))
-			raw.Attachments = []fmsg.AttachmentHeader{{Type: "text/plain;charset=UTF-8", Filename: "note.txt", Size: att.Size, Filepath: att.Filepath}}
-			wire := prepared(t, raw)
-			published := hashOf(t, wire)
-			bodySize, attSize := raw.Size, raw.Attachments[0].Size
-			raw.Size = wire.Size
-			raw.Attachments[0].Size = wire.Attachments[0].Size
-			var header []byte
-			if retainHeader {
-				header = wire.Encode()
-			}
-			id := putOld(t, db, raw, nil, published, header)
-			if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-				t.Fatal(err)
-			}
-			var gotBody, gotAtt uint32
-			var hash, snapshot []byte
-			if err := db.QueryRow(`SELECT size,sha256,wire_message FROM msg WHERE id=$1`, id).Scan(&gotBody, &hash, &snapshot); err != nil {
-				t.Fatal(err)
-			}
-			if err := db.QueryRow(`SELECT filesize FROM msg_attachment WHERE msg_id=$1`, id).Scan(&gotAtt); err != nil {
-				t.Fatal(err)
-			}
-			if gotBody != bodySize || gotAtt != attSize || !bytes.Equal(hash, published) {
-				t.Fatal("expanded metadata or published hash changed")
-			}
-			if _, err := fmsg.UnmarshalPrepared(snapshot, published); err != nil {
-				t.Fatal(err)
-			}
-		})
-	}
-}
-
-func TestMigrationHistoricalLocalLinkAndUnhashedReceivedBatch(t *testing.T) {
-	db := previousStore(t)
-	parentRaw := rawMessage(t, "example.com", "parent")
-	parentHash := hashOf(t, parentRaw)
-	parent := putOld(t, db, parentRaw, nil, parentHash, nil)
-	childRaw := rawMessage(t, "example.com", "child")
-	historicalHash := hashOf(t, childRaw)
-	child := putOld(t, db, childRaw, parent, historicalHash, nil)
-	batch := putBatch(t, db, parent, oldBatch{from: fmsg.Address{User: "bob", Domain: "example.org"}, time: 1300, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}})
-	if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-		t.Fatal(err)
-	}
-	var pid int64
-	var hash, parentSHA, data []byte
-	if err := db.QueryRow(`SELECT pid,sha256,psha256,wire_message FROM msg WHERE id=$1`, child).Scan(&pid, &hash, &parentSHA, &data); err != nil {
-		t.Fatal(err)
-	}
-	if pid != parent || !bytes.Equal(hash, historicalHash) || !bytes.Equal(parentSHA, parentHash) {
-		t.Fatal("historical identity or local thread link changed")
-	}
-	h, err := fmsg.UnmarshalPrepared(data, historicalHash)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if h.Flags&fmsg.FlagHasPid != 0 {
-		t.Fatal("changed historical root-form identity")
-	}
-	if err := db.QueryRow(`SELECT sha256,wire_message FROM msg_add_to_batch WHERE id=$1`, batch).Scan(&hash, &data); err != nil {
-		t.Fatal(err)
-	}
-	if _, err := fmsg.UnmarshalPrepared(data, hash); err != nil {
-		t.Fatal(err)
-	}
-}
-
-func TestMigrationPreservesEarlyLocalNotesAndLiteralRecipients(t *testing.T) {
-	db := previousStore(t)
-	raw := rawMessage(t, "example.com", "local note")
-	raw.To = nil
-	hash := hashOf(t, raw)
-	note := putOld(t, db, raw, nil, hash, nil)
-	bad := rawMessage(t, "example.com", "literal recipient")
-	bad.To = []fmsg.Address{{User: "alice", Domain: "example.org,@bob@example.com"}}
-	literal := putOld(t, db, bad, nil, nil, nil)
-	child := rawMessage(t, "example.com", "local child")
-	childHash := hashOf(t, child)
-	parentRaw := rawMessage(t, "example.com", "unhashed local parent")
-	parent := putOld(t, db, parentRaw, nil, nil, nil)
-	reply := putOld(t, db, child, parent, childHash, nil)
-	if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-		t.Fatal(err)
-	}
-	for _, id := range []int64{note, literal, reply} {
-		var got, data []byte
-		if err := db.QueryRow(`SELECT sha256,wire_message FROM msg WHERE id=$1`, id).Scan(&got, &data); err != nil {
-			t.Fatal(err)
-		}
-		h, err := fmsg.UnmarshalPrepared(data, got)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if id == note && (!bytes.Equal(got, hash) || len(h.To) != 0) {
-			t.Fatal("local note changed")
-		}
-		if id == literal && h.To[0].ToString() != bad.To[0].ToString() {
-			t.Fatal("literal recipient changed")
-		}
-		if id == reply && !bytes.Equal(got, childHash) {
-			t.Fatal("historical child identity changed")
-		}
-	}
-}
-
-func TestMigrationRecoversRecordedBatchWireTime(t *testing.T) {
-	db := previousStore(t)
-	raw := rawMessage(t, "example.org", "forwarded")
-	base := prepared(t, raw)
-	canonical := hashOf(t, base)
-	b := oldBatch{from: raw.From, time: 1300, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}}
-	wire := batchHeader(base, canonical, b)
-	hash := hashOf(t, wire)
-	raw.Timestamp = 1300
-	id := putOld(t, db, raw, nil, canonical, wire.Encode())
-	b.time = 1302
-	bid := putBatch(t, db, id, b)
-	if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-		t.Fatal(err)
-	}
-	var stamp float64
-	var got []byte
-	if err := db.QueryRow(`SELECT time_added,sha256 FROM msg_add_to_batch WHERE id=$1`, bid).Scan(&stamp, &got); err != nil {
-		t.Fatal(err)
-	}
-	if stamp != 1300 || !bytes.Equal(got, hash) {
-		t.Fatal("recorded wire identity not restored")
-	}
-}
-
-func TestMigrationPreservesEarlyHashWithBatchFields(t *testing.T) {
-	db := previousStore(t)
-	parent := rawMessage(t, "example.com", "parent")
-	parentID := putOld(t, db, parent, nil, nil, nil)
-	raw := rawMessage(t, "example.com", "reply")
-	raw.Topic = ""
-	b := oldBatch{from: raw.From, to: []fmsg.Address{{User: "carol", Domain: "remote.example"}}, time: raw.Timestamp}
-	old := raw.Clone()
-	old.Flags = fmsg.FlagHasAddTo
-	old.AddToFrom, old.AddTo = &b.from, b.to
-	hash := hashOf(t, old)
-	id := putOld(t, db, raw, parentID, hash, nil)
-	putBatch(t, db, id, b)
-	if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil {
-		t.Fatal(err)
-	}
-	var got, snapshot []byte
-	if err := db.QueryRow(`SELECT sha256,wire_message FROM msg WHERE id=$1`, id).Scan(&got, &snapshot); err != nil {
-		t.Fatal(err)
-	}
-	if !bytes.Equal(got, hash) {
-		t.Fatal("changed historical identity")
-	}
-	if _, err := fmsg.UnmarshalPrepared(snapshot, hash); err != nil {
-		t.Fatal(err)
-	}
-}
diff --git a/cmd/fmsg-backfill/testdata/previous.sql b/cmd/fmsg-backfill/testdata/previous.sql
deleted file mode 100644
index 2116073..0000000
--- a/cmd/fmsg-backfill/testdata/previous.sql
+++ /dev/null
@@ -1,389 +0,0 @@
-/****************************************************************
- *
- * PostgreSQL database objects data definition for fmsgd
- *
- * This script is IDEMPOTENT: every statement is safe to re-run
- * (create table/index if not exists, alter table add column if
- * not exists, create or replace function, drop trigger if exists
- * before create trigger). Migrating an existing database is
- * therefore just re-running the whole script, e.g.:
- *
- *   psql -d fmsgd -v ON_ERROR_STOP=1 -f dd.sql
- *
- * Keep it that way: add new objects and columns only with
- * idempotent statements, and name indexes explicitly to match
- * PostgreSQL's default generated names so indexes that already
- * exist unnamed on live databases are recognised, not duplicated.
- *
- ****************************************************************/
-
--- database with encoding UTF8 should already be created and connected
-
-create table if not exists msg (
-    id            	bigserial       	primary key,
-	version			int					not null,
-    pid           	bigint          	references msg (id),
-	no_reply		boolean				not null default false,
-	is_important	boolean				not null default false,
-	is_deflate		boolean				not null default false,
-	is_terminal		boolean				not null default false, -- SPEC §3 bit 6: leaf message, nothing may reference it via pid
-    time_sent     	double precision,             -- time sending host recieved message for sending, message timestamp field, NULL means message not ready for sending i.e. draft
-    from_addr     	varchar(255)    	not null,
-    topic         	varchar(255)    	not null,
-    type          	varchar(255)    	not null,
-    sha256        	bytea           	unique,
-    psha256       	bytea,
-	size			int					not null, -- spec allows uint32 but we don't enforced by FMSG_MAX_MSG_SIZE
-    filepath      	text            	not null,
-    wire_header   	bytea                         -- received messages: the exact wire header bytes (fields 1-13), so any hash can always be faithfully recomputed (SPEC §11); null for locally-authored messages
-);
-create index if not exists msg_lower_idx on msg ((lower(from_addr)));
-alter table msg add column if not exists wire_header bytea; -- upgrade path for databases created before this column
-alter table msg add column if not exists is_terminal boolean not null default false; -- upgrade path (SPEC v0.6.0)
-
-create table if not exists msg_to (
-	id				bigserial			primary key,
-	msg_id			bigint				not null references msg (id),
-	addr			varchar(255)		not null,
-    time_delivered  double precision,   -- if sending, time sending host recieved delivery confirmation, if receiving, time successfully received message
-    time_last_attempt double precision, -- only used when sending, time of last delivery attempt if failed; otherwise null
-    time_read       double precision,   -- time recipient read the message; null if unread
-    response_code   smallint,		    -- when sending, response code of last delivery attempt if failed; when receiving, the per-recipient code this host responded, or a negative local sentinel (-1 attempt got no response, retryable; -2 recorded from an exchange, another host's delivery)
-    attempt_count   int             not null default 0, -- number of failed delivery attempts; used for exponential back-off
-	unique (msg_id, addr)
-);
-create index if not exists msg_to_lower_idx on msg_to ((lower(addr)));
-
--- Each add-to delivery for a shared message is one batch: a single sender
--- (add_to_from) added a set of recipients at a point in time. Storing batches
--- separately lets readers reconstruct who added which recipients and when,
--- which a single flat recipient list cannot preserve (SPEC §12). A batch's
--- identity is its message hash (sha256), which covers the batch's time: the
--- same addresses re-issued at a new time are a distinct batch, not a
--- duplicate (SPEC §11/§12). sha256 is null for rows recorded before this
--- column existed and for locally originated batches not yet hashed.
-create table if not exists msg_add_to_batch (
-	id				bigserial			primary key,
-	msg_id			bigint				not null references msg (id),
-	add_to_from		varchar(255)		not null,           -- sender that added this batch's recipients
-	time_added		double precision	not null,           -- the batch message's wire time field (for locally originated batches, when the batch was created)
-	sha256			bytea                                   -- batch message hash: the batch's identity (SPEC §11)
-);
-alter table msg_add_to_batch add column if not exists sha256 bytea;
-create index if not exists msg_add_to_batch_msg_id_idx on msg_add_to_batch (msg_id);
-
-create table if not exists msg_add_to (
-	id				bigserial			primary key,
-	msg_id			bigint				not null references msg (id),
-	batch_id		bigint				not null references msg_add_to_batch (id), -- batch this recipient was added in
-	addr			varchar(255)		not null,
-    time_delivered  double precision,   -- if sending, time sending host recieved delivery confirmation, if receiving, time successfully received message
-    time_last_attempt double precision, -- only used when sending, time of last delivery attempt if failed; otherwise null
-    time_read       double precision,   -- time recipient read the message; null if unread
-    response_code   smallint,		    -- when sending, response code of last delivery attempt if failed; when receiving, the per-recipient code this host responded, or a negative local sentinel (-1 attempt got no response, retryable; -2 recorded from an exchange, another host's delivery)
-    attempt_count   int             not null default 0, -- number of failed delivery attempts; used for exponential back-off
-	unique (batch_id, addr)
-);
--- An address is unique within a batch, not across batches: distinct batches
--- may re-add the same address (each batch is its own sibling branch, SPEC
--- §12). Migrate existing databases off the old per-message constraint.
-alter table msg_add_to drop constraint if exists msg_add_to_msg_id_addr_key;
-create unique index if not exists msg_add_to_batch_id_addr_key on msg_add_to (batch_id, addr);
-create index if not exists msg_add_to_lower_idx on msg_add_to ((lower(addr)));
-create index if not exists msg_add_to_batch_id_idx on msg_add_to (batch_id);
-
-create table if not exists msg_attachment (
-    msg_id        	bigint          references msg (id),
-    position      	smallint        not null default 0,
-    flags         	smallint        not null default 0,
-    type          	varchar(255)    not null default 'application/octet-stream',
-    filename      	varchar(255)    not null,
-    filesize      	int             not null, 
-    filepath      	text			not null,
-    primary key (msg_id, filename)
-);
-
--- keep protocol parent hash populated for locally-created replies that set
--- the relational parent id. A reply cannot reference a draft parent or a
--- terminal parent (SPEC v0.6.0 §3: a Sending Host must not transmit a reply
--- to a terminal message, so refuse to create one), and any explicit psha256
--- must match the referenced parent's sha256.
-create or replace function populate_msg_psha256_from_pid() returns trigger as $$
-declare
-    parent_time_sent double precision;
-    parent_sha256 bytea;
-    parent_is_terminal boolean;
-begin
-    if NEW.pid is null then
-        return NEW;
-    end if;
-
-    select parent.time_sent, parent.sha256, parent.is_terminal
-    into parent_time_sent, parent_sha256, parent_is_terminal
-    from msg parent
-    where parent.id = NEW.pid;
-
-    if not found then
-        raise exception 'parent message % does not exist', NEW.pid;
-    end if;
-
-    if parent_time_sent is null then
-        raise exception 'cannot set pid %: parent message is a draft', NEW.pid;
-    end if;
-
-    if parent_is_terminal then
-        raise exception 'cannot set pid %: parent message is terminal', NEW.pid;
-    end if;
-
-    if parent_sha256 is null or octet_length(parent_sha256) = 0 then
-        -- parent was delivered locally only and has no sha256 yet; psha256 cannot be populated
-        return NEW;
-    end if;
-
-    if NEW.psha256 is null or octet_length(NEW.psha256) = 0 then
-        NEW.psha256 = parent_sha256;
-    elsif NEW.psha256 <> parent_sha256 then
-        -- a reply may reference one of the parent's add-to batch messages by
-        -- its batch hash (SPEC §12); the relational parent is the shared row
-        if not exists (
-            select 1 from msg_add_to_batch b
-            where b.msg_id = NEW.pid and b.sha256 = NEW.psha256
-        ) then
-            raise exception 'psha256 does not match parent message % sha256 or any of its add-to batch hashes', NEW.pid;
-        end if;
-    end if;
-
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_msg_populate_psha256 on msg;
-create trigger trg_msg_populate_psha256
-    before insert or update of pid, psha256 on msg
-    for each row execute function populate_msg_psha256_from_pid();
-
--- recipients cannot be added to a terminal message (SPEC §12): refuse to
--- create a batch for one, so the sender never has such a unit to transmit.
-create or replace function prevent_add_to_terminal_msg() returns trigger as $$
-begin
-    if exists (select 1 from msg where id = NEW.msg_id and is_terminal) then
-        raise exception 'cannot add recipients to message %: it is terminal', NEW.msg_id;
-    end if;
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_msg_add_to_batch_terminal on msg_add_to_batch;
-create trigger trg_msg_add_to_batch_terminal
-    before insert on msg_add_to_batch
-    for each row execute function prevent_add_to_terminal_msg();
-
--- once a message has replies, it must remain referenceable by protocol hash.
-create or replace function prevent_referenced_msg_from_becoming_unreferenceable() returns trigger as $$
-begin
-    if exists (select 1 from msg child where child.pid = NEW.id) then
-        if NEW.time_sent is null then
-            raise exception 'cannot make message % a draft: it has replies', NEW.id;
-        end if;
-
-        if OLD.sha256 is not null and (NEW.sha256 is null or octet_length(NEW.sha256) = 0) then
-            raise exception 'cannot clear sha256 for message %: it has replies', NEW.id;
-        end if;
-
-        if OLD.sha256 is distinct from NEW.sha256 then
-            raise exception 'cannot change sha256 for message %: it has replies', NEW.id;
-        end if;
-    end if;
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_msg_prevent_unreferenceable_parent on msg;
-create trigger trg_msg_prevent_unreferenceable_parent
-    before update of time_sent, sha256 on msg
-    for each row execute function prevent_referenced_msg_from_becoming_unreferenceable();
-
--- Notify the sender's outgoing worker (channel new_msg_to) whenever new
--- delivery work appears. One function serves all three triggers, dispatching
--- on the table it fired for:
---   * msg               -- a draft message transitions to sent (time_sent set
---                          for the first time); notify every recipient.
---   * msg_to/msg_add_to -- a recipient row is inserted against an already-sent
---                          message (recipients added via add-to after the
---                          message was sent, including a freshly inserted
---                          message whose recipient rows follow in the same
---                          transaction); notify that recipient.
--- The payload is advisory only: the worker re-polls fully on any wake-up.
-create or replace function notify_msg_sent() returns trigger as $$
-begin
-    if TG_TABLE_NAME = 'msg' then
-        if OLD.time_sent is null and NEW.time_sent is not null then
-            perform pg_notify('new_msg_to', NEW.id::text || ',' || addr)
-            from msg_to where msg_id = NEW.id;
-
-            perform pg_notify('new_msg_to', NEW.id::text || ',' || addr)
-            from msg_add_to where msg_id = NEW.id;
-        end if;
-    elsif NEW.time_delivered is null then
-        perform pg_notify('new_msg_to', NEW.msg_id::text || ',' || NEW.addr)
-        from msg where id = NEW.msg_id and time_sent is not null;
-    end if;
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_msg_to_insert on msg_to;
-create trigger trg_msg_to_insert
-    after insert on msg_to
-    for each row execute function notify_msg_sent();
-
-drop trigger if exists trg_msg_add_to_insert on msg_add_to;
-create trigger trg_msg_add_to_insert
-    after insert on msg_add_to
-    for each row execute function notify_msg_sent();
-
-drop trigger if exists trg_msg_sent on msg;
-create trigger trg_msg_sent
-    after update on msg
-    for each row execute function notify_msg_sent();
-
--- Notify listeners (channel new_msg) that a message has become sent/arrived:
--- time_sent set for the first time, on insert (e.g. a message received from a
--- remote host) or update (a local draft being sent). Unlike new_msg_to this
--- fires regardless of recipient domain, so push-notification listeners can wake
--- without polling. Payload is "<msg id>,<addr>", one notification per recipient
--- -- the listener checks addr against its currently-subscribed clients and only
--- fetches message detail for those that are connected.
---
--- This is a DEFERRABLE constraint trigger so it runs at COMMIT: on insert the
--- msg row is written before its msg_to/msg_add_to rows (FK ordering), so a
--- plain row trigger would see no recipients. At commit every recipient row in
--- the transaction is visible.
-create or replace function notify_new_msg() returns trigger as $$
-begin
-    if (TG_OP = 'INSERT' and NEW.time_sent is not null) or
-       (TG_OP = 'UPDATE' and OLD.time_sent is null and NEW.time_sent is not null) then
-        perform pg_notify('new_msg', NEW.id::text || ',' || addr)
-        from msg_to where msg_id = NEW.id;
-
-        perform pg_notify('new_msg', NEW.id::text || ',' || addr)
-        from msg_add_to where msg_id = NEW.id;
-    end if;
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_new_msg on msg;
-create constraint trigger trg_new_msg
-    after insert or update on msg
-    deferrable initially deferred
-    for each row execute function notify_new_msg();
-
--- Notify the sender (channel delivered) once a recipient's delivery is
--- confirmed, so the sender's UI can unlock replying without a manual reload.
--- Fires on the NULL -> non-NULL transition of time_delivered, which happens
--- once per recipient row regardless of who performs the UPDATE (fmsgd's own
--- remote delivery, its local-domain delivery, or fmsg-webapi's same-domain
--- delivery) -- triggering on the tables rather than the call site covers all
--- of them. Payload is "<msg id>,<from_addr>", the same shape as new_msg's
--- payload but with the sender's address instead of the recipient's, since
--- it's the sender whose UI needs to react. Unlike trg_new_msg this does not
--- need to be deferred: the msg row referenced by msg_id already exists (FK)
--- by the time msg_to/msg_add_to is updated.
-create or replace function notify_delivered() returns trigger as $$
-begin
-    perform pg_notify('delivered', NEW.msg_id::text || ',' || m.from_addr)
-    from msg m where m.id = NEW.msg_id;
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_msg_to_delivered on msg_to;
-create trigger trg_msg_to_delivered
-    after update of time_delivered on msg_to
-    for each row
-    when (OLD.time_delivered is null and NEW.time_delivered is not null)
-    execute function notify_delivered();
-
-drop trigger if exists trg_msg_add_to_delivered on msg_add_to;
-create trigger trg_msg_add_to_delivered
-    after update of time_delivered on msg_add_to
-    for each row
-    when (OLD.time_delivered is null and NEW.time_delivered is not null)
-    execute function notify_delivered();
-
--- Sender-side state for add-to participant notification (SPEC §10.2): an
--- add-to message is sent to every participant domain of the message being
--- added to -- the domains of from and every to address as well as the new
--- recipients' -- so all participants learn recipients were added, not only
--- the domains hosting the new recipients. Domains hosting a recipient of the
--- batch itself learn through normal recipient delivery; every other
--- participant domain gets one row here per batch and receives the add-to as
--- a notification-only exchange completing at code 11. Rows are created by
--- the Web API when recipients are added through it (the local domain itself
--- needs no row -- this database is its record).
-create table if not exists msg_add_to_notify (
-    id                bigserial        primary key,
-    batch_id          bigint           not null references msg_add_to_batch (id),
-    domain            varchar(255)     not null,
-    time_notified     double precision,   -- time remote host acknowledged the batch; null means pending
-    time_last_attempt double precision,   -- time of last failed attempt; drives exponential back-off
-    response_code     smallint,           -- response code of last attempt
-    attempt_count     int              not null default 0,
-    unique (batch_id, domain)
-);
-
--- Wake the sender's outgoing worker (channel new_msg_to) for a pending
--- participant notification, mirroring notify_msg_sent for recipient rows.
--- The payload is advisory only: the worker re-polls fully on any wake-up.
-create or replace function notify_add_to_notify_pending() returns trigger as $$
-begin
-    perform pg_notify('new_msg_to', b.msg_id::text || ',' || NEW.domain)
-    from msg_add_to_batch b
-    inner join msg m on m.id = b.msg_id
-    where b.id = NEW.batch_id and m.time_sent is not null;
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_msg_add_to_notify_insert on msg_add_to_notify;
-create trigger trg_msg_add_to_notify_insert
-    after insert on msg_add_to_notify
-    for each row execute function notify_add_to_notify_pending();
-
--- Notify listeners (channel recipients_added) that an add-to batch was
--- recorded against a sent message, so existing participants' clients learn of
--- the new recipients without polling. Fires wherever a batch is recorded --
--- added locally through the Web API or received from a remote host -- because
--- both paths insert a msg_add_to_batch row. Payload is "<msg id>,<addr>", one
--- notification per participant (from, every msg_to and every msg_add_to
--- address, including the new batch's own recipients, who have no other
--- realtime event for a message that was sent before they were added); the
--- listener checks addr against its currently-connected clients, exactly as
--- new_msg. Like trg_new_msg this is a deferred constraint trigger: the
--- batch's own msg_add_to rows are inserted after the batch row, so only at
--- commit is the full recipient set visible.
-create or replace function notify_recipients_added() returns trigger as $$
-begin
-    if not exists (select 1 from msg where id = NEW.msg_id and time_sent is not null) then
-        return NEW;
-    end if;
-
-    perform pg_notify('recipients_added', NEW.msg_id::text || ',' || from_addr)
-    from msg where id = NEW.msg_id;
-
-    perform pg_notify('recipients_added', NEW.msg_id::text || ',' || addr)
-    from msg_to where msg_id = NEW.msg_id;
-
-    perform pg_notify('recipients_added', NEW.msg_id::text || ',' || addr)
-    from msg_add_to where msg_id = NEW.msg_id;
-
-    return NEW;
-end;
-$$ language plpgsql;
-
-drop trigger if exists trg_recipients_added on msg_add_to_batch;
-create constraint trigger trg_recipients_added
-    after insert on msg_add_to_batch
-    deferrable initially deferred
-    for each row execute function notify_recipients_added();
diff --git a/cmd/fmsg-backfill/wire.go b/cmd/fmsg-backfill/wire.go
deleted file mode 100644
index 6fe6934..0000000
--- a/cmd/fmsg-backfill/wire.go
+++ /dev/null
@@ -1,228 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"compress/zlib"
-	"encoding/binary"
-	"fmt"
-	"io"
-	"os"
-	"path/filepath"
-
-	"github.com/markmnl/fmsgd/pkg/fmsg"
-)
-
-// Decode the old wire_header column offline, without network parser state or
-// present-day timestamp limits. Round-tripping must reproduce every byte.
-func decodeHeader(data []byte) (*fmsg.Header, error) {
-	d := headerReader{r: bytes.NewReader(data)}
-	h := &fmsg.Header{Version: d.byte(), Flags: d.byte()}
-	if h.Flags&fmsg.FlagHasPid != 0 {
-		h.Pid = d.take(32)
-	}
-	h.From = d.address()
-	h.To = d.addresses()
-	if h.Flags&fmsg.FlagHasAddTo != 0 {
-		a := d.address()
-		h.AddToFrom = &a
-		h.AddTo = d.addresses()
-	}
-	d.number(&h.Timestamp)
-	if h.Flags&fmsg.FlagHasPid == 0 {
-		h.Topic = d.text()
-	}
-	h.Type, h.TypeID = d.mediaType(h.Flags&fmsg.FlagCommonType != 0)
-	d.number(&h.Size)
-	if h.Flags&fmsg.FlagDeflate != 0 {
-		d.number(&h.ExpandedSize)
-	}
-	count := d.byte()
-	for i := 0; i < int(count); i++ {
-		a := fmsg.AttachmentHeader{Flags: d.byte()}
-		a.Type, a.TypeID = d.mediaType(a.Flags&1 != 0)
-		a.Filename = d.text()
-		d.number(&a.Size)
-		if a.Flags&2 != 0 {
-			d.number(&a.ExpandedSize)
-		}
-		if a.Flags&^uint8(3) != 0 {
-			d.err = fmt.Errorf("reserved attachment flags in stored header")
-		}
-		h.Attachments = append(h.Attachments, a)
-	}
-	if d.err != nil {
-		return nil, fmt.Errorf("invalid stored wire header: %w", d.err)
-	}
-	if h.Version != 1 || h.Flags&128 != 0 || len(h.To) == 0 ||
-		(h.Flags&fmsg.FlagHasAddTo != 0 && (len(h.AddTo) == 0 || len(h.Pid) != 32)) ||
-		d.r.Len() != 0 || !bytes.Equal(h.Encode(), data) {
-		return nil, fmt.Errorf("stored wire header does not round-trip")
-	}
-	return h, nil
-}
-
-type headerReader struct {
-	r   *bytes.Reader
-	err error
-}
-
-func (d *headerReader) take(n int) []byte {
-	b := make([]byte, n)
-	if d.err == nil {
-		_, d.err = io.ReadFull(d.r, b)
-	}
-	return b
-}
-func (d *headerReader) byte() byte   { return d.take(1)[0] }
-func (d *headerReader) text() string { return string(d.take(int(d.byte()))) }
-func (d *headerReader) number(v any) {
-	if d.err == nil {
-		d.err = binary.Read(d.r, binary.LittleEndian, v)
-	}
-}
-func (d *headerReader) address() fmsg.Address {
-	raw := d.text()
-	if d.err != nil {
-		return fmsg.Address{}
-	}
-	a, err := address(raw)
-	if err != nil {
-		d.err = err
-	}
-	return a
-}
-func (d *headerReader) addresses() []fmsg.Address {
-	n := int(d.byte())
-	list := make([]fmsg.Address, n)
-	for i := range list {
-		list[i] = d.address()
-	}
-	return list
-}
-func (d *headerReader) mediaType(common bool) (string, uint8) {
-	if !common {
-		return d.text(), 0
-	}
-	id := d.byte()
-	typ, ok := fmsg.GetCommonMediaType(id)
-	if !ok {
-		d.err = fmt.Errorf("unknown common media type %d", id)
-	}
-	return typ, id
-}
-
-// Previous receivers retained expanded API files and the wire header, but not
-// compressed payload files. Recreate a valid stream of the declared size. The
-// protocol hashes expanded bytes, so compressed bytes need not be identical;
-// the header (including wire size) and expanded bytes must be identical.
-func (m *migration) restore(wire, raw *fmsg.Header) (*fmsg.Header, error) {
-	h := wire.Clone()
-	var temps []string
-	defer func() {
-		for _, path := range temps {
-			_ = os.Remove(path)
-		}
-	}()
-	part := func(path string, rawSize, size, expanded uint32, compressed bool) (string, error) {
-		info, err := os.Stat(path)
-		if err != nil {
-			return "", err
-		}
-		want := size
-		if compressed {
-			want = expanded
-		}
-		if !info.Mode().IsRegular() || info.Size() != int64(want) || rawSize != want {
-			return "", fmt.Errorf("stored payload length differs from wire header: %s", path)
-		}
-		if !compressed {
-			return path, nil
-		}
-		p, err := restoreCompressed(path, size)
-		if err == nil {
-			temps = append(temps, p)
-		}
-		return p, err
-	}
-	var err error
-	h.Filepath, err = part(raw.Filepath, raw.Size, h.Size, h.ExpandedSize, h.Flags&fmsg.FlagDeflate != 0)
-	if err != nil {
-		return nil, err
-	}
-	if len(h.Attachments) != len(raw.Attachments) {
-		return nil, fmt.Errorf("stored attachments differ from wire header")
-	}
-	for i := range h.Attachments {
-		a := &h.Attachments[i]
-		found := false
-		for _, r := range raw.Attachments {
-			if r.Filename == a.Filename {
-				a.Filepath, err = part(r.Filepath, r.Size, a.Size, a.ExpandedSize, a.Flags&2 != 0)
-				if err != nil {
-					return nil, err
-				}
-				found = true
-				break
-			}
-		}
-		if !found {
-			return nil, fmt.Errorf("missing attachment %s", a.Filename)
-		}
-	}
-	h, dir, err := fmsg.Preserve(h, filepath.Dir(raw.Filepath))
-	if err == nil {
-		m.files = append(m.files, dir)
-	}
-	return h, err
-}
-
-func restoreCompressed(path string, size uint32) (string, error) {
-	in, err := os.Open(path)
-	if err != nil {
-		return "", err
-	}
-	defer in.Close()
-	out, err := os.CreateTemp("", "fmsg-backfill-zlib-*")
-	if err != nil {
-		return "", err
-	}
-	keep := false
-	defer func() {
-		out.Close()
-		if !keep {
-			os.Remove(out.Name())
-		}
-	}()
-	for _, level := range []int{zlib.DefaultCompression, 1, 9, 0, zlib.HuffmanOnly, 2, 3, 4, 5, 7, 8} {
-		if _, err = in.Seek(0, io.SeekStart); err != nil {
-			return "", err
-		}
-		if err = out.Truncate(0); err != nil {
-			return "", err
-		}
-		if _, err = out.Seek(0, io.SeekStart); err != nil {
-			return "", err
-		}
-		zw, err := zlib.NewWriterLevel(out, level)
-		if err != nil {
-			return "", err
-		}
-		_, copyErr := io.Copy(zw, in)
-		err = zw.Close()
-		if copyErr != nil {
-			return "", copyErr
-		}
-		if err != nil {
-			return "", err
-		}
-		n, err := out.Seek(0, io.SeekCurrent)
-		if err != nil {
-			return "", err
-		}
-		if n == int64(size) {
-			keep = true
-			return out.Name(), nil
-		}
-	}
-	return "", fmt.Errorf("cannot reconstruct %d-byte compressed representation of %s; restore original wire payload before upgrading", size, path)
-}
diff --git a/dd.sql b/dd.sql
index a3d640d..4ed5605 100644
--- a/dd.sql
+++ b/dd.sql
@@ -1,8 +1,14 @@
--- PostgreSQL bootstrap schema for a new, empty fmsg message database.
--- Existing installations use the standalone fmsg-backfill binary before
--- starting this version. This file is not an upgrade script.
+-- PostgreSQL data definition for fmsgd.
+--
+-- This script is IDEMPOTENT: every statement is safe to re-run (create
+-- table/index if not exists, create or replace function, drop trigger if
+-- exists before create trigger), so deploys re-run the whole script:
+--
+--   psql -d fmsgd -v ON_ERROR_STOP=1 -f dd.sql
+--
+-- Keep it that way.
 
-create table msg (
+create table if not exists msg (
     id            	bigserial       	primary key,
 	version			int					not null,
     pid           	bigint          	references msg (id),
@@ -21,9 +27,9 @@ create table msg (
     wire_header   	bytea,                        -- exact protocol header (fields 1-13)
     wire_message    jsonb                         -- durable original wire representation; null for drafts or originals received only through add-to
 );
-create index msg_lower_idx on msg ((lower(from_addr)));
+create index if not exists msg_lower_idx on msg ((lower(from_addr)));
 
-create table msg_to (
+create table if not exists msg_to (
 	id				bigserial			primary key,
 	msg_id			bigint				not null references msg (id),
 	addr			varchar(255)		not null,
@@ -34,7 +40,7 @@ create table msg_to (
     attempt_count   int             not null default 0, -- number of failed delivery attempts; used for exponential back-off
 	unique (msg_id, addr)
 );
-create index msg_to_lower_idx on msg_to ((lower(addr)));
+create index if not exists msg_to_lower_idx on msg_to ((lower(addr)));
 
 -- Each add-to delivery for a shared message is one batch: a single sender
 -- (add_to_from) added a set of recipients at a point in time. Storing batches
@@ -43,7 +49,7 @@ create index msg_to_lower_idx on msg_to ((lower(addr)));
 -- identity is its message hash (sha256), which covers the batch's time: the
 -- same addresses re-issued at a new time are a distinct batch, not a
 -- duplicate (SPEC §11/§12). Batches of a draft finalize when it is sent.
-create table msg_add_to_batch (
+create table if not exists msg_add_to_batch (
 	id				bigserial			primary key,
 	msg_id			bigint				not null references msg (id),
 	add_to_from		varchar(255)		not null,           -- sender that added this batch's recipients
@@ -51,9 +57,9 @@ create table msg_add_to_batch (
 	sha256			bytea,                                  -- finalized batch identity (SPEC §11)
     wire_message    jsonb                                   -- durable batch wire representation
 );
-create index msg_add_to_batch_msg_id_idx on msg_add_to_batch (msg_id);
+create index if not exists msg_add_to_batch_msg_id_idx on msg_add_to_batch (msg_id);
 
-create table msg_add_to (
+create table if not exists msg_add_to (
 	id				bigserial			primary key,
 	msg_id			bigint				not null references msg (id),
 	batch_id		bigint				not null references msg_add_to_batch (id), -- batch this recipient was added in
@@ -65,10 +71,10 @@ create table msg_add_to (
     attempt_count   int             not null default 0, -- number of failed delivery attempts; used for exponential back-off
 	unique (batch_id, addr)
 );
-create index msg_add_to_lower_idx on msg_add_to ((lower(addr)));
-create index msg_add_to_batch_id_idx on msg_add_to (batch_id);
+create index if not exists msg_add_to_lower_idx on msg_add_to ((lower(addr)));
+create index if not exists msg_add_to_batch_id_idx on msg_add_to (batch_id);
 
-create table msg_attachment (
+create table if not exists msg_attachment (
     msg_id        	bigint          references msg (id),
     position      	smallint        not null default 0,
     flags         	smallint        not null default 0,
@@ -89,7 +95,7 @@ create table msg_attachment (
 -- a notification-only exchange completing at code 11. Rows are created by
 -- the Web API when recipients are added through it (the local domain itself
 -- needs no row -- this database is its record).
-create table msg_add_to_notify (
+create table if not exists msg_add_to_notify (
     id                bigserial        primary key,
     batch_id          bigint           not null references msg_add_to_batch (id),
     domain            varchar(255)     not null,
@@ -100,8 +106,8 @@ create table msg_add_to_notify (
     unique (batch_id, domain)
 );
 
-create index msg_add_to_batch_sha256_idx on msg_add_to_batch (sha256) where sha256 is not null;
-create index msg_pid_idx on msg (pid) where pid is not null;
+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;
 
 -- Functions and triggers.
 
@@ -110,7 +116,7 @@ create index msg_pid_idx on msg (pid) where pid is not null;
 -- terminal parent (SPEC v0.6.0 §3: a Sending Host must not transmit a reply
 -- to a terminal message, so refuse to create one), and any explicit psha256
 -- must match the referenced parent's sha256.
-create function populate_msg_psha256_from_pid() returns trigger as $$
+create or replace function populate_msg_psha256_from_pid() returns trigger as $$
 declare
     parent_time_sent double precision;
     parent_sha256 bytea;
@@ -158,13 +164,14 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_msg_populate_psha256 on msg;
 create trigger trg_msg_populate_psha256
     before insert or update of pid, psha256 on msg
     for each row execute function populate_msg_psha256_from_pid();
 
 -- recipients cannot be added to a terminal message (SPEC §12): refuse to
 -- create a batch for one, so the sender never has such a unit to transmit.
-create function prevent_add_to_terminal_msg() returns trigger as $$
+create or replace function prevent_add_to_terminal_msg() returns trigger as $$
 begin
     if exists (select 1 from msg where id = NEW.msg_id and is_terminal) then
         raise exception 'cannot add recipients to message %: it is terminal', NEW.msg_id;
@@ -173,6 +180,7 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_msg_add_to_batch_terminal on msg_add_to_batch;
 create trigger trg_msg_add_to_batch_terminal
     before insert on msg_add_to_batch
     for each row execute function prevent_add_to_terminal_msg();
@@ -188,7 +196,7 @@ create trigger trg_msg_add_to_batch_terminal
 --                          message whose recipient rows follow in the same
 --                          transaction); notify that recipient.
 -- The payload is advisory only: the worker re-polls fully on any wake-up.
-create function notify_msg_sent() returns trigger as $$
+create or replace function notify_msg_sent() returns trigger as $$
 begin
     if TG_TABLE_NAME = 'msg' then
         if OLD.time_sent is null and NEW.time_sent is not null then
@@ -206,14 +214,17 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_msg_to_insert on msg_to;
 create trigger trg_msg_to_insert
     after insert on msg_to
     for each row execute function notify_msg_sent();
 
+drop trigger if exists trg_msg_add_to_insert on msg_add_to;
 create trigger trg_msg_add_to_insert
     after insert on msg_add_to
     for each row execute function notify_msg_sent();
 
+drop trigger if exists trg_msg_sent on msg;
 create trigger trg_msg_sent
     after update on msg
     for each row execute function notify_msg_sent();
@@ -230,7 +241,7 @@ create trigger trg_msg_sent
 -- msg row is written before its msg_to/msg_add_to rows (FK ordering), so a
 -- plain row trigger would see no recipients. At commit every recipient row in
 -- the transaction is visible.
-create function notify_new_msg() returns trigger as $$
+create or replace function notify_new_msg() returns trigger as $$
 begin
     if (TG_OP = 'INSERT' and NEW.time_sent is not null) or
        (TG_OP = 'UPDATE' and OLD.time_sent is null and NEW.time_sent is not null) then
@@ -244,6 +255,7 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_new_msg on msg;
 create constraint trigger trg_new_msg
     after insert or update on msg
     deferrable initially deferred
@@ -260,7 +272,7 @@ create constraint trigger trg_new_msg
 -- it's the sender whose UI needs to react. Unlike trg_new_msg this does not
 -- need to be deferred: the msg row referenced by msg_id already exists (FK)
 -- by the time msg_to/msg_add_to is updated.
-create function notify_delivered() returns trigger as $$
+create or replace function notify_delivered() returns trigger as $$
 begin
     perform pg_notify('delivered', NEW.msg_id::text || ',' || m.from_addr)
     from msg m where m.id = NEW.msg_id;
@@ -268,12 +280,14 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_msg_to_delivered on msg_to;
 create trigger trg_msg_to_delivered
     after update of time_delivered on msg_to
     for each row
     when (OLD.time_delivered is null and NEW.time_delivered is not null)
     execute function notify_delivered();
 
+drop trigger if exists trg_msg_add_to_delivered on msg_add_to;
 create trigger trg_msg_add_to_delivered
     after update of time_delivered on msg_add_to
     for each row
@@ -283,7 +297,7 @@ create trigger trg_msg_add_to_delivered
 -- Wake the sender's outgoing worker (channel new_msg_to) for a pending
 -- participant notification, mirroring notify_msg_sent for recipient rows.
 -- The payload is advisory only: the worker re-polls fully on any wake-up.
-create function notify_add_to_notify_pending() returns trigger as $$
+create or replace function notify_add_to_notify_pending() returns trigger as $$
 begin
     perform pg_notify('new_msg_to', b.msg_id::text || ',' || NEW.domain)
     from msg_add_to_batch b
@@ -293,6 +307,7 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_msg_add_to_notify_insert on msg_add_to_notify;
 create trigger trg_msg_add_to_notify_insert
     after insert on msg_add_to_notify
     for each row execute function notify_add_to_notify_pending();
@@ -309,7 +324,7 @@ create trigger trg_msg_add_to_notify_insert
 -- new_msg. Like trg_new_msg this is a deferred constraint trigger: the
 -- batch's own msg_add_to rows are inserted after the batch row, so only at
 -- commit is the full recipient set visible.
-create function notify_recipients_added() returns trigger as $$
+create or replace function notify_recipients_added() returns trigger as $$
 begin
     if not exists (select 1 from msg where id = NEW.msg_id and time_sent is not null) then
         return NEW;
@@ -328,6 +343,7 @@ begin
 end;
 $$ language plpgsql;
 
+drop trigger if exists trg_recipients_added on msg_add_to_batch;
 create constraint trigger trg_recipients_added
     after insert on msg_add_to_batch
     deferrable initially deferred
@@ -335,7 +351,7 @@ create constraint trigger trg_recipients_added
 
 -- Sent protocol fields are immutable. Relational pid links and delivery/read
 -- metadata remain bookkeeping and may change.
-create function protect_msg_identity() returns trigger as $$
+create or replace function protect_msg_identity() returns trigger as $$
 begin
     if OLD.time_sent is not null and
        row(NEW.time_sent,NEW.sha256,NEW.version,NEW.psha256,NEW.no_reply,
@@ -350,11 +366,12 @@ begin
     return NEW;
 end;
 $$ language plpgsql;
+drop trigger if exists trg_msg_identity on msg;
 create trigger trg_msg_identity before update on msg for each row execute function protect_msg_identity();
 
 -- Validate after all rows in the transaction have been assembled. An original
 -- first received via add-to has its payload representation on the received batch.
-create function require_sent_msg_hash() returns trigger as $$
+create or replace function require_sent_msg_hash() returns trigger as $$
 begin
     if exists (select 1 from msg m where m.id=NEW.id and m.time_sent is not null
                and (m.sha256 is null or octet_length(m.sha256) <> 32 or
@@ -371,10 +388,11 @@ begin
     return null;
 end;
 $$ language plpgsql;
+drop trigger if exists trg_msg_require_hash on msg;
 create constraint trigger trg_msg_require_hash after insert or update on msg
     deferrable initially deferred for each row execute function require_sent_msg_hash();
 
-create function require_sent_batch_hash() returns trigger as $$
+create or replace function require_sent_batch_hash() returns trigger as $$
 begin
     if exists (select 1 from msg_add_to_batch b join msg m on m.id=b.msg_id
                where b.id=NEW.id and m.time_sent is not null
@@ -384,10 +402,11 @@ begin
     return null;
 end;
 $$ language plpgsql;
+drop trigger if exists trg_batch_require_hash on msg_add_to_batch;
 create constraint trigger trg_batch_require_hash after insert or update on msg_add_to_batch
     deferrable initially deferred for each row execute function require_sent_batch_hash();
 
-create function protect_msg_parts() returns trigger as $$
+create or replace function protect_msg_parts() returns trigger as $$
 declare
     message_id bigint;
     frozen boolean;
@@ -420,11 +439,14 @@ begin
 end;
 $$ language plpgsql;
 -- AFTER INSERT allows an ON CONFLICT DO NOTHING receipt to remain a no-op.
+drop trigger if exists trg_msg_to_content on msg_to;
 create trigger trg_msg_to_content after insert or update or delete on msg_to for each row execute function protect_msg_parts();
+drop trigger if exists trg_msg_attachment_content on msg_attachment;
 create trigger trg_msg_attachment_content after insert or update or delete on msg_attachment for each row execute function protect_msg_parts();
+drop trigger if exists trg_msg_add_to_content on msg_add_to;
 create trigger trg_msg_add_to_content after insert or update or delete on msg_add_to for each row execute function protect_msg_parts();
 
-create function protect_batch_identity() returns trigger as $$
+create or replace function protect_batch_identity() returns trigger as $$
 begin
     if OLD.sha256 is not null and
        row(NEW.msg_id,NEW.add_to_from,NEW.time_added,NEW.sha256,NEW.wire_message)
@@ -434,4 +456,5 @@ begin
     return NEW;
 end;
 $$ language plpgsql;
+drop trigger if exists trg_batch_identity on msg_add_to_batch;
 create trigger trg_batch_identity before update on msg_add_to_batch for each row execute function protect_batch_identity();
diff --git a/schema.go b/schema.go
deleted file mode 100644
index 5f4fca0..0000000
--- a/schema.go
+++ /dev/null
@@ -1,9 +0,0 @@
-// Package fmsgd exposes the bootstrap schema for offline maintenance tools.
-package fmsgd
-
-import _ "embed"
-
-// Schema is the schema for a new message database. The daemon does not apply it.
-//
-//go:embed dd.sql
-var Schema string
diff --git a/temp.sql b/temp.sql
deleted file mode 100644
index 7c80a8b..0000000
--- a/temp.sql
+++ /dev/null
@@ -1,52 +0,0 @@
-/****************************************************************
- *
- * One-off migration: add-to batch provenance
- *
- * Brings an existing fmsgd database up to the current dd.sql schema:
- * introduces msg_add_to_batch / msg_add_to.batch_id and removes the legacy
- * single msg.add_to_from column. A flat msg_add_to list with one set-once
- * add_to_from could not preserve which sender added which recipients, nor
- * when -- this backfills one synthetic batch per message so that history is
- * (approximately) recoverable.
- *
- * Safe to re-run: every step is guarded, so a second run (or a database that
- * already has the new shape) is a no-op. Run msg_add_to_batch's CREATE from
- * dd.sql first if it does not yet exist
- *
- * Apply with: psql "<conn>" -f temp.sql
- *
- ****************************************************************/
-
--- add column before its index so an existing msg_add_to (created before
--- batch_id) is altered first.
-alter table msg_add_to add column if not exists batch_id bigint references msg_add_to_batch (id);
-create index if not exists msg_add_to_batch_id_idx on msg_add_to (batch_id);
-
-do $$
-begin
-    -- Backfill one synthetic batch per message that already has add-to
-    -- recipients, sourcing the sender from the legacy column, then drop it.
-    if exists (
-        select 1 from information_schema.columns
-        where table_name = 'msg' and column_name = 'add_to_from'
-    ) then
-        insert into msg_add_to_batch (msg_id, add_to_from, time_added)
-        select m.id, coalesce(nullif(m.add_to_from, ''), m.from_addr), coalesce(m.time_sent, 0)
-        from msg m
-        where exists (
-            select 1 from msg_add_to a where a.msg_id = m.id and a.batch_id is null
-        );
-
-        update msg_add_to a
-        set batch_id = b.id
-        from msg_add_to_batch b
-        where a.batch_id is null and b.msg_id = a.msg_id;
-
-        alter table msg drop column add_to_from;
-    end if;
-
-    -- Tighten the FK to match dd.sql once every recipient is linked.
-    if not exists (select 1 from msg_add_to where batch_id is null) then
-        alter table msg_add_to alter column batch_id set not null;
-    end if;
-end $$;