Recover historical message encodings in standalone backfill

cmd/fmsg-backfill/migrate.go

@@ -39,7 +39,7 @@ type oldBatch struct {
 }
 
 func address(s string) (fmsg.Address, error) {
-	p := strings.Split(s, "@")
+	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)
 	}
@@ -176,14 +176,19 @@ func (m *migration) message(id int64) error {
 			return fmt.Errorf("parent %d: %w", *s.pid, err)
 		}
 		if len(s.h.Pid) == 0 {
-			if len(s.hash) > 0 {
-				return fmt.Errorf("hashed reply has no parent hash; cannot change its identity")
-			}
 			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
@@ -231,7 +236,7 @@ func (m *migration) message(id int64) error {
 			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)
+			base, err = m.reconstruct(s.h, s.hash, batches)
 		}
 	}
 	if err != nil {
@@ -259,6 +264,32 @@ func (m *migration) message(id int64) error {
 	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)
@@ -268,10 +299,6 @@ func (m *migration) message(id int64) error {
 		if len(b.prepared) > 0 {
 			h, err = fmsg.UnmarshalPrepared(b.prepared, b.hash)
 		} else {
-			if len(b.hash) == 0 && !strings.EqualFold(b.from.Domain, m.domain) &&
-				(received == nil || !bytes.Equal(h.Encode(), received.Encode())) {
-				return fmt.Errorf("remote batch %d has no published hash or exact header", b.id)
-			}
 			h, err = selectTypes(h, b.hash)
 		}
 		if err != nil {
@@ -295,6 +322,14 @@ func (m *migration) message(id int64) error {
 	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
 }
@@ -310,12 +345,20 @@ func batchHeader(base *fmsg.Header, hash []byte, b oldBatch) *fmsg.Header {
 // 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) (*fmsg.Header, error) {
-	h, dir, err := fmsg.Prepare(raw)
+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
 	}
-	if chosen, err := selectTypes(h, expected); err == nil {
+	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
 	}
@@ -324,7 +367,7 @@ func (m *migration) reconstruct(raw *fmsg.Header, expected []byte) (*fmsg.Header
 	if err != nil {
 		return nil, err
 	}
-	chosen, err := selectTypes(h, expected)
+	chosen, err := selectHistorical(h, expected, batches)
 	if err != nil {
 		_ = os.RemoveAll(dir)
 		return nil, err
@@ -333,6 +376,41 @@ func (m *migration) reconstruct(raw *fmsg.Header, expected []byte) (*fmsg.Header
 	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()
@@ -388,3 +466,28 @@ func bytesOrNull(b []byte) any {
 	}
 	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
+}

cmd/fmsg-backfill/migrate_test.go

@@ -326,3 +326,167 @@ func TestDecodeHeaderRejectsTruncation(t *testing.T) {
 		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)
+	}
+}