commit 51e3819c205eb9ea79421f82e00fc70dd21bb2e1
parent 176e95a8855422c7e2c42275abdf6f1ed5068848
Author: MTRNord <MTRNord@users.noreply.github.com>
Date: Sat, 4 Apr 2026 15:11:21 +0200
various improvements across the board
Signed-off-by: MTRNord <MTRNord@users.noreply.github.com>
Diffstat:
15 files changed, 2013 insertions(+), 1744 deletions(-)
diff --git a/apps/talos_cluster/matrix-backup/backup-cronjob.yaml b/apps/talos_cluster/matrix-backup/backup-cronjob.yaml
@@ -37,6 +37,7 @@ spec:
cpu: 100m
memory: 128Mi
limits:
+ cpu: 500m
memory: 512Mi
env:
- name: HOMESERVER
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/.gitignore b/apps/talos_cluster/matrix-backup/backup-tool/.gitignore
@@ -1,2 +1,2 @@
-decrypt/
+recovery/
decrypted/
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/accountdata.go b/apps/talos_cluster/matrix-backup/backup-tool/accountdata.go
@@ -0,0 +1,154 @@
+// accountdata.go — backup of global account data, per-room account data, and
+// the initial profile snapshot. All three use a retry-on-next-run pattern:
+// failures are logged as warnings so the next CronJob run retries automatically.
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "log/slog"
+ "net/url"
+
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/event"
+)
+
+// backupAccountData saves a snapshot of all global account-data events to S3.
+// The strategy is overwrite-every-run so the file is always current; a failed
+// write leaves the previous version in place and the next run retries.
+//
+// S3 keys:
+//
+// {prefix}/account-data-latest.json.age
+// {prefix}/account-data-{dateStr}.json.age
+func backupAccountData(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync, prefix string) {
+ accountData := make(map[string]json.RawMessage)
+
+ // Seed from the sync response (present on first sync and when changed).
+ for _, ev := range syncResp.AccountData.Events {
+ if ev.Content.VeryRaw != nil {
+ accountData[ev.Type.Type] = ev.Content.VeryRaw
+ }
+ }
+
+ // Always fetch these key types directly from the API so they're present even
+ // on incremental syncs where account data hasn't changed.
+ for _, t := range []string{"m.push_rules", "m.ignored_user_list", "m.direct"} {
+ if _, ok := accountData[t]; !ok {
+ var raw json.RawMessage
+ if err := getAccountData(ctx, client, t, &raw); err == nil && raw != nil {
+ accountData[t] = raw
+ }
+ }
+ }
+
+ if len(accountData) == 0 {
+ slog.Info("No account data to save", "prefix", prefix)
+ return
+ }
+
+ data, err := json.MarshalIndent(accountData, "", " ")
+ if err != nil {
+ slog.Warn("Failed to marshal account data", "error", err)
+ return
+ }
+ if err := s3PutAge(ctx, prefix+"/account-data-latest.json", data); err != nil {
+ slog.Warn("Failed to save account data", "prefix", prefix, "error", err)
+ return
+ }
+ if err := s3PutAge(ctx, prefix+"/account-data-"+dateStr+".json", data); err != nil {
+ slog.Warn("Failed to save dated account data", "prefix", prefix, "error", err)
+ }
+ slog.Info("Account data saved", "prefix", prefix, "types", len(accountData))
+}
+
+// backupRoomAccountData saves per-room account-data (room tags, read markers,
+// etc.) to S3 using a merge pattern so rooms absent from this sync retain their
+// last-known values.
+//
+// S3 key:
+//
+// {prefix}/room-account-data-latest.json.age
+func backupRoomAccountData(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync, prefix string) {
+ // Load existing data (merge pattern — rooms not in this sync are preserved).
+ existing := make(map[string]map[string]json.RawMessage)
+ if raw, err := getDecryptedAgeFromS3(ctx, prefix+"/room-account-data-latest.json.age"); err != nil {
+ slog.Warn("Could not load previous room account data", "error", err)
+ } else if raw != nil {
+ if err := json.Unmarshal(raw, &existing); err != nil {
+ slog.Warn("Previous room account data is corrupt, starting fresh", "error", err)
+ }
+ }
+
+ for roomID, joinedRoom := range syncResp.Rooms.Join {
+ roomData := existing[string(roomID)]
+ if roomData == nil {
+ roomData = make(map[string]json.RawMessage)
+ }
+
+ // Collect all account-data types present in the sync delta.
+ for _, ev := range joinedRoom.AccountData.Events {
+ if ev.Content.VeryRaw != nil {
+ roomData[ev.Type.Type] = ev.Content.VeryRaw
+ }
+ }
+
+ // Always fetch m.room.tag directly so it's present even without a delta.
+ tagPath := "/_matrix/client/v3/user/" +
+ url.PathEscape(client.UserID.String()) +
+ "/rooms/" +
+ url.PathEscape(string(roomID)) +
+ "/tags"
+ var tags json.RawMessage
+ if err := matrixGetJSON(ctx, client, tagPath, &tags); err == nil && tags != nil {
+ roomData[event.AccountDataRoomTags.Type] = tags
+ }
+
+ existing[string(roomID)] = roomData
+ }
+
+ data, err := json.MarshalIndent(existing, "", " ")
+ if err != nil {
+ slog.Warn("Failed to marshal room account data", "error", err)
+ return
+ }
+ if err := s3PutAge(ctx, prefix+"/room-account-data-latest.json", data); err != nil {
+ slog.Warn("Failed to save room account data", "prefix", prefix, "error", err)
+ return
+ }
+ slog.Info("Room account data saved", "prefix", prefix, "rooms", len(existing))
+}
+
+// backupProfileSnapshot saves the user's current displayname and avatar_url to
+// S3 on the first run. Subsequent runs skip it (write-once semantics); if the
+// write failed on a previous run the file is absent and this run will retry.
+//
+// S3 key:
+//
+// {prefix}/profile-snapshot.json.age
+func backupProfileSnapshot(ctx context.Context, client *mautrix.Client, prefix string) {
+ snapshotKey := prefix + "/profile-snapshot.json.age"
+ if s3Exists(ctx, snapshotKey) {
+ return
+ }
+ var profile struct {
+ Displayname string `json:"displayname"`
+ AvatarURL string `json:"avatar_url"`
+ }
+ path := "/_matrix/client/v3/profile/" + url.PathEscape(client.UserID.String())
+ if err := matrixGetJSON(ctx, client, path, &profile); err != nil {
+ slog.Warn("Failed to fetch profile snapshot", "error", err)
+ return
+ }
+ data, err := json.MarshalIndent(profile, "", " ")
+ if err != nil {
+ slog.Warn("Failed to marshal profile snapshot", "error", err)
+ return
+ }
+ // s3PutAge appends ".age" — writes to prefix+"/profile-snapshot.json.age".
+ if err := s3PutAge(ctx, prefix+"/profile-snapshot.json", data); err != nil {
+ slog.Warn("Failed to save profile snapshot", "prefix", prefix, "error", err)
+ return
+ }
+ slog.Info("Profile snapshot saved", "prefix", prefix, "displayname", profile.Displayname)
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/backup.go b/apps/talos_cluster/matrix-backup/backup-tool/backup.go
@@ -0,0 +1,154 @@
+// backup.go — per-account backup orchestration.
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "log/slog"
+
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/event"
+ "maunium.net/go/mautrix/id"
+)
+
+// backupAccount runs the full backup pipeline for a single Matrix account:
+// restore crypto state, sync, export E2EE keys, save room list, paginate
+// history, save account data, then persist state back to S3.
+func backupAccount(ctx context.Context, acc accountCfg) error {
+ slog.Info("=== Backing up account ===", "user_id", acc.UserID)
+
+ storeS3Key := acc.Prefix + "/crypto-store.tar.gz"
+ sessionS3Key := acc.Prefix + "/session.json"
+ syncTokenS3Key := acc.Prefix + "/sync-token.json"
+
+ if err := downloadStore(ctx, acc.StoreDir, storeS3Key); err != nil {
+ slog.Warn("Could not restore store", "error", err)
+ }
+
+ client, err := mautrix.NewClient(homeserver, "", "")
+ if err != nil {
+ return err
+ }
+
+ if _, err := ensureSession(ctx, client, acc, sessionS3Key); err != nil {
+ return err
+ }
+
+ var syncToken struct {
+ NextBatch string `json:"next_batch"`
+ }
+ _ = s3GetJSON(ctx, syncTokenS3Key, &syncToken)
+
+ slog.Info("Syncing", "since", syncToken.NextBatch)
+ syncResp, err := client.SyncRequest(ctx, 60000, syncToken.NextBatch, "", true, event.PresenceUnavailable)
+ if err != nil {
+ return err
+ }
+ if data, err := json.Marshal(map[string]string{"next_batch": syncResp.NextBatch}); err == nil {
+ if err := s3Put(ctx, syncTokenS3Key, data, "application/json"); err != nil {
+ slog.Warn("Failed to save sync token", "error", err)
+ }
+ }
+ slog.Info("Sync done", "rooms", len(syncResp.Rooms.Join))
+
+ var sessions megolmSessions
+ if acc.SSSSKey != "" {
+ sessions, err = fetchAndExportKeyBackup(ctx, client, acc.SSSSKey, acc.Prefix)
+ if err != nil {
+ slog.Warn("Key backup export failed", "error", err)
+ }
+ }
+
+ if err := saveRoomList(ctx, client, syncResp, acc.Prefix); err != nil {
+ slog.Warn("Room list failed", "error", err)
+ }
+
+ // Back up account-data, per-room account data, and profile snapshot.
+ backupAccountData(ctx, client, syncResp, acc.Prefix)
+ backupRoomAccountData(ctx, client, syncResp, acc.Prefix)
+ backupProfileSnapshot(ctx, client, acc.Prefix)
+
+ // Build the DM room set: m.direct API + 2-member heuristic from stored list.
+ dmRooms := getDMRooms(ctx, client, syncResp)
+ if raw, err := getDecryptedAgeFromS3(ctx, acc.Prefix+"/rooms-latest.json.age"); err == nil && raw != nil {
+ var storedRooms []roomEntry
+ if json.Unmarshal(raw, &storedRooms) == nil {
+ for _, r := range storedRooms {
+ if r.Type == "dm" {
+ dmRooms[id.RoomID(r.RoomID)] = true
+ }
+ }
+ }
+ }
+
+ // Paginate history for all rooms present in the sync delta.
+ for roomID, joinedRoom := range syncResp.Rooms.Join {
+ isDM := dmRooms[roomID]
+ if err := paginateRoom(ctx, client, roomID, acc.Prefix, isDM, joinedRoom.Timeline.PrevBatch, syncResp.NextBatch, sessions); err != nil {
+ slog.Warn("History error", "room_id", roomID, "error", err)
+ }
+ }
+
+ // Catch-up: DM rooms absent from this sync delta (no new events) may still
+ // need their history backfilled. Use the stored rooms list as the authoritative
+ // DM source, falling back to the API-based dmRooms map.
+ catchupDMs := make(map[id.RoomID]bool)
+ for roomID := range dmRooms {
+ catchupDMs[roomID] = true
+ }
+ if raw, err := getDecryptedAgeFromS3(ctx, acc.Prefix+"/rooms-latest.json.age"); err != nil {
+ slog.Warn("DM catch-up: failed to read rooms list", "error", err)
+ } else if raw == nil {
+ slog.Warn("DM catch-up: rooms list not found or ageIdentity not set")
+ } else {
+ var storedRooms []roomEntry
+ if err := json.Unmarshal(raw, &storedRooms); err != nil {
+ slog.Warn("DM catch-up: failed to parse rooms list", "error", err)
+ } else {
+ slog.Info("DM catch-up: loaded rooms list", "total_rooms", len(storedRooms))
+ for _, r := range storedRooms {
+ if r.Type == "dm" {
+ catchupDMs[id.RoomID(r.RoomID)] = true
+ }
+ }
+ }
+ }
+ slog.Info("DM catch-up: checking rooms", "total_dm_rooms", len(catchupDMs), "in_sync", len(syncResp.Rooms.Join))
+ for roomID := range catchupDMs {
+ if _, inSync := syncResp.Rooms.Join[roomID]; inSync {
+ continue // already handled above
+ }
+ safeKey := roomSafeKey(roomID)
+ dmHistoryDoneKey := acc.Prefix + "/history-cursor/" + safeKey + ".dm-done"
+ dmDone, dmDoneErr := s3Get(ctx, dmHistoryDoneKey)
+ if dmDoneErr != nil {
+ slog.Warn("DM catch-up: error checking dm-done marker", "room_id", roomID, "error", dmDoneErr)
+ }
+ if dmDone != nil {
+ continue // already backfilled
+ }
+ cursorKey := acc.Prefix + "/history-cursor/" + safeKey + ".json"
+ var cursor struct {
+ Token string `json:"token"`
+ }
+ if err := s3GetJSON(ctx, cursorKey, &cursor); err != nil {
+ slog.Warn("DM catch-up: error reading cursor", "room_id", roomID, "error", err)
+ continue
+ }
+ if cursor.Token == "" {
+ slog.Info("DM catch-up: no cursor yet, skipping", "room_id", roomID)
+ continue
+ }
+ slog.Info("DM catch-up: backfilling room", "room_id", roomID, "cursor", cursor.Token)
+ if err := paginateRoom(ctx, client, roomID, acc.Prefix, true, cursor.Token, syncResp.NextBatch, sessions); err != nil {
+ slog.Warn("DM catch-up error", "room_id", roomID, "error", err)
+ }
+ }
+
+ if err := uploadStore(ctx, acc.StoreDir, storeS3Key); err != nil {
+ slog.Warn("Could not upload store", "error", err)
+ }
+
+ slog.Info("Done", "user_id", acc.UserID)
+ return nil
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/config.go b/apps/talos_cluster/matrix-backup/backup-tool/config.go
@@ -0,0 +1,114 @@
+// config.go — global configuration, account definitions, and age key initialisation.
+package main
+
+import (
+ "fmt"
+ "log/slog"
+ "os"
+ "strings"
+ "time"
+
+ "filippo.io/age"
+ "maunium.net/go/mautrix/id"
+)
+
+// ourServerName is used to decide whose media and avatars to download.
+// Override with S3_SERVER_NAME if needed.
+var ourServerName = envOrDefault("SERVER_NAME", "mtrnord.blog")
+
+type accountCfg struct {
+ UserID id.UserID
+ Password string
+ SSSSKey string
+ Prefix string
+ StoreDir string
+}
+
+var (
+ homeserver = mustEnv("HOMESERVER")
+ s3Endpoint = mustEnv("S3_ENDPOINT")
+ s3BucketName = mustEnv("S3_BUCKET")
+ s3AccessKey = mustEnv("S3_ACCESS_KEY")
+ s3SecretKey = mustEnv("S3_SECRET_KEY")
+ s3Region = envOrDefault("S3_REGION", "hel1")
+ keyExportPass = mustEnv("KEY_EXPORT_PASSPHRASE")
+ dateStr = time.Now().UTC().Format("20060102")
+ runStr = time.Now().UTC().Format("20060102-150405")
+
+ accounts = []accountCfg{
+ {
+ UserID: "@mtrnord:mtrnord.blog",
+ Password: os.Getenv("MTRNORD_PASSWORD"),
+ SSSSKey: os.Getenv("MTRNORD_SSSS_KEY"),
+ Prefix: "mtrnord",
+ StoreDir: "/data/crypto/mtrnord",
+ },
+ {
+ UserID: "@lexi:mtrnord.blog",
+ Password: os.Getenv("LEXI_PASSWORD"),
+ SSSSKey: os.Getenv("LEXI_SSSS_KEY"),
+ Prefix: "lexi",
+ StoreDir: "/data/crypto/lexi",
+ },
+ }
+
+ // ageRecipients holds the parsed public keys used to encrypt history files.
+ ageRecipients []age.Recipient
+
+ // ageIdentity is the optional private key used to decrypt files written by
+ // this tool (loaded from AGE_PRIVATE_KEY). Allows reading back previously
+ // encrypted objects without storing plain copies in S3.
+ ageIdentity age.Identity
+)
+
+func mustEnv(key string) string {
+ v := os.Getenv(key)
+ if v == "" {
+ slog.Error("Required environment variable not set", "key", key)
+ os.Exit(1)
+ }
+ return v
+}
+
+func envOrDefault(key, def string) string {
+ if v := os.Getenv(key); v != "" {
+ return v
+ }
+ return def
+}
+
+// initAgeRecipients parses a comma-separated list of age public keys.
+func initAgeRecipients(s string) error {
+ for _, raw := range strings.Split(s, ",") {
+ raw = strings.TrimSpace(raw)
+ if raw == "" {
+ continue
+ }
+ r, err := age.ParseX25519Recipient(raw)
+ if err != nil {
+ return fmt.Errorf("invalid age recipient %q: %w", raw, err)
+ }
+ ageRecipients = append(ageRecipients, r)
+ }
+ if len(ageRecipients) == 0 {
+ return fmt.Errorf("AGE_RECIPIENTS is empty — at least one public key is required")
+ }
+ return nil
+}
+
+// initAgeIdentity parses an AGE-SECRET-KEY-1… line for decrypting previously
+// written .age objects (e.g. the room list, session token).
+func initAgeIdentity(privKey string) error {
+ privKey = strings.TrimSpace(privKey)
+ if privKey == "" {
+ return nil
+ }
+ ids, err := age.ParseIdentities(strings.NewReader(privKey))
+ if err != nil {
+ return fmt.Errorf("parse age identity: %w", err)
+ }
+ if len(ids) > 0 {
+ ageIdentity = ids[0]
+ }
+ return nil
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/events.go b/apps/talos_cluster/matrix-backup/backup-tool/events.go
@@ -0,0 +1,167 @@
+// events.go — per-event processing: media download, profile-change recording,
+// and the historyEvent record type shared by both pagination paths.
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "log/slog"
+ "strings"
+
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/event"
+ "maunium.net/go/mautrix/id"
+)
+
+// historyEvent is the on-disk representation of a single room event stored
+// in JSONL files under prefix/history/<safeKey>/<run>.jsonl.age.
+type historyEvent struct {
+ EventID string `json:"event_id"`
+ Sender string `json:"sender"`
+ Type string `json:"type"`
+ Timestamp int64 `json:"origin_server_ts"`
+ Content json.RawMessage `json:"content"`
+ Encrypted bool `json:"encrypted,omitempty"`
+}
+
+func eventToRecord(ev *event.Event, wasEncrypted bool) historyEvent {
+ return historyEvent{
+ EventID: string(ev.ID),
+ Sender: string(ev.Sender),
+ Type: ev.Type.Type,
+ Timestamp: ev.Timestamp,
+ Content: ev.Content.VeryRaw,
+ Encrypted: wasEncrypted,
+ }
+}
+
+// roomSafeKey converts a room ID into a string safe for use as an S3 key
+// component by replacing '/' and ':' with '_'.
+func roomSafeKey(roomID id.RoomID) string {
+ return strings.NewReplacer("/", "_", ":", "_").Replace(string(roomID))
+}
+
+// processEvent downloads media and records profile changes for a single event.
+// The event must already be decrypted before this is called.
+func processEvent(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string, isDM bool) {
+ if ev.Type == event.EventEncrypted {
+ return
+ }
+
+ if ev.Type == event.EventMessage {
+ var content struct {
+ MsgType string `json:"msgtype"`
+ URL string `json:"url"`
+ File json.RawMessage `json:"file"`
+ }
+ if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil {
+ mediaURL := content.URL
+ if mediaURL == "" && content.File != nil {
+ var f struct {
+ URL string `json:"url"`
+ }
+ if json.Unmarshal(content.File, &f) == nil {
+ mediaURL = f.URL
+ }
+ }
+ if mediaMsgTypes[content.MsgType] && mediaURL != "" {
+ if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) {
+ if err := downloadAndStoreMedia(ctx, client, mediaURL, prefix, "media"); err != nil {
+ slog.Warn("Media download failed", "url", mediaURL, "error", err)
+ }
+ }
+ }
+ }
+ }
+
+ if ev.Type == event.EventSticker {
+ var content struct {
+ URL string `json:"url"`
+ File json.RawMessage `json:"file"`
+ }
+ if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil {
+ mediaURL := content.URL
+ if mediaURL == "" && content.File != nil {
+ var f struct {
+ URL string `json:"url"`
+ }
+ if json.Unmarshal(content.File, &f) == nil {
+ mediaURL = f.URL
+ }
+ }
+ if mediaURL != "" {
+ if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) {
+ if err := downloadAndStoreMedia(ctx, client, mediaURL, prefix, "media"); err != nil {
+ slog.Warn("Sticker download failed", "url", mediaURL, "error", err)
+ }
+ }
+ }
+ }
+ }
+
+ if ev.Type == event.StateMember {
+ recordProfileUpdate(ctx, client, ev, roomID, prefix)
+ }
+}
+
+type profileUpdateRecord struct {
+ TS int64 `json:"ts"`
+ EventID string `json:"event_id"`
+ UserID string `json:"user_id"`
+ RoomID string `json:"room_id"`
+ DisplayNameOld string `json:"displayname_old,omitempty"`
+ DisplayNameNew string `json:"displayname_new,omitempty"`
+ AvatarOld string `json:"avatar_old,omitempty"`
+ AvatarNew string `json:"avatar_new,omitempty"`
+}
+
+// recordProfileUpdate appends a JSONL entry to S3 when a member event shows a
+// display-name or avatar change for a local user; also downloads the new avatar.
+func recordProfileUpdate(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string) {
+ if ev.StateKey == nil {
+ return
+ }
+ stateKey := *ev.StateKey
+ if !strings.HasSuffix(stateKey, ":"+ourServerName) {
+ return
+ }
+ var content, prevContent struct {
+ Displayname string `json:"displayname"`
+ AvatarURL string `json:"avatar_url"`
+ }
+ if err := json.Unmarshal(ev.Content.VeryRaw, &content); err != nil {
+ return
+ }
+ if ev.Unsigned.PrevContent != nil {
+ _ = json.Unmarshal(ev.Unsigned.PrevContent.VeryRaw, &prevContent)
+ }
+ if content.Displayname == prevContent.Displayname && content.AvatarURL == prevContent.AvatarURL {
+ return
+ }
+ rec := profileUpdateRecord{
+ TS: ev.Timestamp,
+ EventID: string(ev.ID),
+ UserID: stateKey,
+ RoomID: string(roomID),
+ }
+ if content.Displayname != prevContent.Displayname {
+ rec.DisplayNameOld = prevContent.Displayname
+ rec.DisplayNameNew = content.Displayname
+ }
+ if content.AvatarURL != prevContent.AvatarURL {
+ rec.AvatarOld = prevContent.AvatarURL
+ rec.AvatarNew = content.AvatarURL
+ if content.AvatarURL != "" {
+ if err := downloadAndStoreMedia(ctx, client, content.AvatarURL, prefix, "avatars"); err != nil {
+ slog.Warn("Avatar download failed", "url", content.AvatarURL, "error", err)
+ }
+ }
+ }
+ line, err := json.Marshal(rec)
+ if err != nil {
+ return
+ }
+ s3Key := prefix + "/profile-updates.jsonl"
+ existing, _ := s3Get(ctx, s3Key)
+ _ = s3Put(ctx, s3Key, append(existing, append(line, '\n')...), "application/x-ndjson")
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/history.go b/apps/talos_cluster/matrix-backup/backup-tool/history.go
@@ -0,0 +1,206 @@
+// history.go — room history pagination: forward incremental sync and DM
+// backward backfill.
+package main
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "log/slog"
+
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/event"
+ "maunium.net/go/mautrix/id"
+)
+
+// paginateRoom fetches room history and writes new events to a per-run
+// age-encrypted JSONL file in S3. For DMs on first run the full history is
+// fetched backwards from the current position before normal forward pagination.
+//
+// syncNextBatch is the nextBatch token from the current sync response. On first
+// run it is saved as the forward cursor so the next run only sees events that
+// arrive after this sync — avoiding re-fetching events already in the sync.
+func paginateRoom(
+ ctx context.Context,
+ client *mautrix.Client,
+ roomID id.RoomID,
+ prefix string,
+ isDM bool,
+ prevBatch string,
+ syncNextBatch string,
+ sessions megolmSessions,
+) error {
+ safeKey := roomSafeKey(roomID)
+ cursorKey := prefix + "/history-cursor/" + safeKey + ".json"
+ dmHistoryDoneKey := prefix + "/history-cursor/" + safeKey + ".dm-done"
+
+ var cursor struct {
+ Token string `json:"token"`
+ }
+ if err := s3GetJSON(ctx, cursorKey, &cursor); err != nil {
+ return err
+ }
+
+ // fetchDMHistory does a full backward walk from fromToken and stores results.
+ // Used on first run for DMs and as catch-up when a room was initially missed.
+ fetchDMHistory := func(fromToken string) {
+ slog.Info("DM: fetching full history", "room_id", roomID, "from_token", fromToken)
+ var messages []historyEvent
+ token := fromToken
+ for {
+ resp, err := client.Messages(ctx, roomID, token, "", mautrix.DirectionBackward, nil, 100)
+ if err != nil {
+ slog.Warn("room_messages error (backward)", "room_id", roomID, "token", token, "error", err)
+ break
+ }
+ slog.Info("DM history page", "room_id", roomID, "chunk_size", len(resp.Chunk), "end", resp.End)
+ if len(resp.Chunk) == 0 {
+ break
+ }
+ for _, ev := range resp.Chunk {
+ wasEncrypted := ev.Type == event.EventEncrypted
+ ev = tryDecryptEvent(ev, sessions)
+ messages = append(messages, eventToRecord(ev, wasEncrypted))
+ processEvent(ctx, client, ev, roomID, prefix, true)
+ }
+ if resp.End == "" || resp.End == token {
+ break
+ }
+ token = resp.End
+ }
+ if len(messages) > 0 {
+ // Reverse so events are stored oldest-first.
+ for i, j := 0, len(messages)-1; i < j; i, j = i+1, j-1 {
+ messages[i], messages[j] = messages[j], messages[i]
+ }
+ histKey := prefix + "/history/" + safeKey + "/" + runStr + "-backfill.jsonl"
+ var buf bytes.Buffer
+ for _, m := range messages {
+ line, err := json.Marshal(m)
+ if err != nil {
+ slog.Warn("Failed to marshal event, skipping", "event_id", m.EventID, "error", err)
+ continue
+ }
+ buf.Write(line)
+ buf.WriteByte('\n')
+ }
+ if err := s3PutAge(ctx, histKey, buf.Bytes()); err != nil {
+ slog.Warn("Failed to upload DM history", "room_id", roomID, "error", err)
+ } else {
+ slog.Info("DM history stored", "room_id", roomID, "events", len(messages))
+ if err := s3Put(ctx, dmHistoryDoneKey, []byte("done"), "text/plain"); err != nil {
+ slog.Warn("Failed to write dm-done marker", "room_id", roomID, "error", err)
+ }
+ }
+ } else {
+ // Nothing to backfill — mark done so we don't retry every run.
+ slog.Info("DM history: no messages found, marking done", "room_id", roomID)
+ if err := s3Put(ctx, dmHistoryDoneKey, []byte("done"), "text/plain"); err != nil {
+ slog.Warn("Failed to write dm-done marker", "room_id", roomID, "error", err)
+ }
+ }
+ }
+
+ if cursor.Token == "" {
+ // First time we've seen this room. For DMs, walk all the way back through
+ // history so we have a complete record from day one.
+ if isDM && prevBatch != "" {
+ fetchDMHistory(prevBatch)
+ }
+ // Save the sync's nextBatch as the forward cursor so the next run only
+ // picks up genuinely new events.
+ if syncNextBatch == "" {
+ syncNextBatch = prevBatch
+ }
+ if syncNextBatch != "" {
+ cursorJSON, err := json.Marshal(map[string]string{"token": syncNextBatch})
+ if err == nil {
+ if err := s3Put(ctx, cursorKey, cursorJSON, "application/json"); err != nil {
+ slog.Warn("Failed to save cursor (first run)", "room_id", roomID, "error", err)
+ }
+ }
+ }
+ return nil
+ }
+
+ // Incremental run: if this is a DM but we never completed a backward history
+ // fetch (e.g. room was missed because it wasn't in m.direct on first run),
+ // do it now using the saved cursor as the backward starting point.
+ // Note: prevBatch may be empty for rooms with no new events in a full_state
+ // sync, so we use cursor.Token directly rather than gating on prevBatch.
+ if isDM {
+ dmDoneData, _ := s3Get(ctx, dmHistoryDoneKey)
+ if dmDoneData == nil {
+ slog.Info("DM catch-up: no history marker found, backfilling", "room_id", roomID)
+ fetchDMHistory(cursor.Token)
+ }
+ }
+
+ // Incremental forward pagination from the saved cursor.
+ var messages []historyEvent
+ nextToken := cursor.Token
+ lastGoodToken := cursor.Token
+ histKey := prefix + "/history/" + safeKey + "/" + runStr + ".jsonl"
+
+ const maxPages = 100
+ pages := 0
+ for range maxPages {
+ pages++
+ resp, err := client.Messages(ctx, roomID, nextToken, "", mautrix.DirectionForward, nil, 100)
+ if err != nil {
+ slog.Warn("room_messages error (forward)", "room_id", roomID, "error", err)
+ break
+ }
+ if len(resp.Chunk) == 0 {
+ break
+ }
+ for _, ev := range resp.Chunk {
+ wasEncrypted := ev.Type == event.EventEncrypted
+ ev = tryDecryptEvent(ev, sessions)
+ messages = append(messages, eventToRecord(ev, wasEncrypted))
+ processEvent(ctx, client, ev, roomID, prefix, isDM)
+ }
+ if resp.End == "" || resp.End == nextToken {
+ lastGoodToken = nextToken
+ nextToken = ""
+ break
+ }
+ lastGoodToken = resp.End
+ nextToken = resp.End
+ }
+ if pages == maxPages && nextToken != "" {
+ slog.Warn("Pagination cap reached — room has more events; next run will continue",
+ "room_id", roomID, "pages", maxPages)
+ lastGoodToken = nextToken
+ }
+
+ if len(messages) > 0 {
+ var buf bytes.Buffer
+ for _, m := range messages {
+ line, err := json.Marshal(m)
+ if err != nil {
+ slog.Warn("Failed to marshal event, skipping", "event_id", m.EventID, "error", err)
+ continue
+ }
+ buf.Write(line)
+ buf.WriteByte('\n')
+ }
+ if err := s3PutAge(ctx, histKey, buf.Bytes()); err != nil {
+ slog.Warn("Failed to upload history", "room_id", roomID, "error", err)
+ }
+ slog.Info("History updated", "room_id", roomID, "new_events", len(messages))
+ }
+
+ // Advance the cursor so the next run never re-processes stored events.
+ if lastGoodToken != cursor.Token {
+ cursorJSON, err := json.Marshal(map[string]string{"token": lastGoodToken})
+ if err != nil {
+ return fmt.Errorf("marshal cursor: %w", err)
+ }
+ if err := s3Put(ctx, cursorKey, cursorJSON, "application/json"); err != nil {
+ slog.Warn("Failed to advance cursor", "room_id", roomID, "error", err)
+ }
+ }
+ return nil
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/keys.go b/apps/talos_cluster/matrix-backup/backup-tool/keys.go
@@ -0,0 +1,327 @@
+// keys.go — SSSS key derivation, Megolm key-backup export, and in-memory
+// session decryption used during history pagination.
+package main
+
+import (
+ "context"
+ "crypto/aes"
+ "crypto/cipher"
+ "crypto/hmac"
+ "crypto/rand"
+ "crypto/sha256"
+ "crypto/sha512"
+ "encoding/base64"
+ "encoding/binary"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log/slog"
+ "net/url"
+ "strings"
+
+ "golang.org/x/crypto/pbkdf2"
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/crypto/backup"
+ "maunium.net/go/mautrix/crypto/goolm/session"
+ "maunium.net/go/mautrix/crypto/ssss"
+ "maunium.net/go/mautrix/event"
+ "maunium.net/go/mautrix/id"
+)
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Types
+// ─────────────────────────────────────────────────────────────────────────────
+
+type defaultKeyEventContent struct {
+ Key string `json:"key"`
+}
+
+// encryptedSecretContent matches the `{"encrypted": {"<keyID>": {...}}}` format.
+type encryptedSecretContent struct {
+ Encrypted map[string]ssss.EncryptedKeyData `json:"encrypted"`
+}
+
+type keyBackupVersionResp struct {
+ Version string `json:"version"`
+ Algorithm string `json:"algorithm"`
+}
+
+type keyBackupRoomSession struct {
+ FirstMessageIndex int `json:"first_message_index"`
+ ForwardedCount int `json:"forwarded_count"`
+ IsVerified bool `json:"is_verified"`
+ SessionData json.RawMessage `json:"session_data"`
+}
+
+type keyBackupRoom struct {
+ Sessions map[string]keyBackupRoomSession `json:"sessions"`
+}
+
+type keyBackupAllRooms struct {
+ Rooms map[id.RoomID]keyBackupRoom `json:"rooms"`
+}
+
+// exportedSessionEntry matches the standard Megolm key export format.
+type exportedSessionEntry struct {
+ Algorithm string `json:"algorithm"`
+ ForwardingCurve25519KeyChain []string `json:"forwarding_curve25519_key_chain"`
+ RoomID string `json:"room_id"`
+ SenderClaimedKeys map[string]string `json:"sender_claimed_keys"`
+ SenderKey string `json:"sender_key"`
+ SessionID string `json:"session_id"`
+ SessionKey string `json:"session_key"`
+}
+
+// megolmSessions maps session IDs to live inbound sessions ready to decrypt.
+type megolmSessions map[id.SessionID]*session.MegolmInboundSession
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Key-backup fetch and export
+// ─────────────────────────────────────────────────────────────────────────────
+
+// fetchAndExportKeyBackup derives the Megolm backup key from SSSS, downloads
+// all room sessions from the server key backup, decrypts them, uploads a JSON
+// session dump and standard key-export file to S3, and returns an in-memory
+// session map for use during history pagination.
+func fetchAndExportKeyBackup(ctx context.Context, client *mautrix.Client, recoveryKeyStr, prefix string) (megolmSessions, error) {
+ var defaultKey defaultKeyEventContent
+ if err := getAccountData(ctx, client, "m.secret_storage.default_key", &defaultKey); err != nil {
+ return nil, fmt.Errorf("get default key: %w", err)
+ }
+ if defaultKey.Key == "" {
+ return nil, fmt.Errorf("m.secret_storage.default_key has no 'key' field")
+ }
+ keyID := defaultKey.Key
+ slog.Info("SSSS default key ID", "key_id", keyID)
+
+ var keyMetadata ssss.KeyMetadata
+ if err := getAccountData(ctx, client, "m.secret_storage.key."+keyID, &keyMetadata); err != nil {
+ return nil, fmt.Errorf("get key metadata: %w", err)
+ }
+
+ // ErrUnverifiableKey means no MAC to check — the derived key is still usable.
+ sssKey, err := keyMetadata.VerifyRecoveryKey(keyID, recoveryKeyStr)
+ if err != nil && !errors.Is(err, ssss.ErrUnverifiableKey) {
+ return nil, fmt.Errorf("verify recovery key: %w", err)
+ }
+ slog.Info("Recovery key verified (or unverifiable but accepted)")
+
+ var backupSecret encryptedSecretContent
+ if err := getAccountData(ctx, client, "m.megolm_backup.v1", &backupSecret); err != nil {
+ return nil, fmt.Errorf("get backup secret: %w", err)
+ }
+ encData, ok := backupSecret.Encrypted[keyID]
+ if !ok {
+ return nil, fmt.Errorf("backup secret not encrypted with key %q", keyID)
+ }
+ backupKeyRaw, err := sssKey.Decrypt("m.megolm_backup.v1", encData)
+ if err != nil {
+ return nil, fmt.Errorf("decrypt backup key from SSSS: %w", err)
+ }
+
+ privateKeyBytes, err := decodeBackupPrivateKey(backupKeyRaw)
+ if err != nil {
+ return nil, fmt.Errorf("decode backup private key: %w", err)
+ }
+ megolmKey, err := backup.MegolmBackupKeyFromBytes(privateKeyBytes)
+ if err != nil {
+ return nil, fmt.Errorf("create megolm backup key: %w", err)
+ }
+ slog.Info("Decoded megolm backup private key from SSSS")
+
+ var backupVersion keyBackupVersionResp
+ if err := matrixGetJSON(ctx, client, "/_matrix/client/v3/room_keys/version", &backupVersion); err != nil {
+ return nil, fmt.Errorf("get backup version: %w", err)
+ }
+ slog.Info("Key backup version", "version", backupVersion.Version)
+
+ var allRooms keyBackupAllRooms
+ if err := matrixGetJSON(ctx, client,
+ "/_matrix/client/v3/room_keys/keys?version="+url.QueryEscape(backupVersion.Version),
+ &allRooms,
+ ); err != nil {
+ return nil, fmt.Errorf("get backup keys: %w", err)
+ }
+ slog.Info("Fetched key backup", "rooms", len(allRooms.Rooms))
+
+ sessions := make(megolmSessions)
+ var exported []exportedSessionEntry
+ decOK, decFail := 0, 0
+ for roomID, room := range allRooms.Rooms {
+ for sessionID, sessionInfo := range room.Sessions {
+ var encSD backup.EncryptedSessionData[backup.MegolmSessionData]
+ if err := json.Unmarshal(sessionInfo.SessionData, &encSD); err != nil {
+ slog.Warn("Failed to parse session_data",
+ "room_id", roomID, "session_id", sessionID, "error", err)
+ decFail++
+ continue
+ }
+ sessionData, err := encSD.Decrypt(megolmKey)
+ if err != nil {
+ slog.Warn("Failed to decrypt session",
+ "room_id", roomID, "session_id", sessionID, "error", err)
+ decFail++
+ continue
+ }
+ if sess, err := session.NewMegolmInboundSessionFromExport([]byte(sessionData.SessionKey)); err == nil {
+ sessions[id.SessionID(sessionID)] = sess
+ } else {
+ slog.Warn("Failed to import Megolm session", "session_id", sessionID, "error", err)
+ }
+ exported = append(exported, exportedSessionEntry{
+ Algorithm: string(sessionData.Algorithm),
+ ForwardingCurve25519KeyChain: sessionData.ForwardingKeyChain,
+ RoomID: string(roomID),
+ SenderClaimedKeys: map[string]string{"ed25519": string(sessionData.SenderClaimedKeys.Ed25519)},
+ SenderKey: string(sessionData.SenderKey),
+ SessionID: sessionID,
+ SessionKey: sessionData.SessionKey,
+ })
+ decOK++
+ }
+ }
+ slog.Info("Session decryption complete", "ok", decOK, "failed", decFail, "in_memory", len(sessions))
+
+ if len(exported) == 0 {
+ slog.Warn("No sessions decrypted — skipping key export")
+ return sessions, nil
+ }
+
+ if rawJSON, err := json.MarshalIndent(exported, "", " "); err == nil {
+ _ = s3PutAge(ctx, prefix+"/backup-sessions-latest.json", rawJSON)
+ _ = s3PutAge(ctx, prefix+"/backup-sessions-"+dateStr+".json", rawJSON)
+ }
+
+ exportData, err := buildMegolmExport(exported, keyExportPass)
+ if err != nil {
+ return sessions, fmt.Errorf("build megolm export: %w", err)
+ }
+ if err := s3PutAge(ctx, prefix+"/crypto-keys-latest.bin", exportData); err != nil {
+ return sessions, err
+ }
+ if err := s3PutAge(ctx, prefix+"/crypto-keys-"+dateStr+".bin", exportData); err != nil {
+ return sessions, err
+ }
+ slog.Info("Exported E2EE keys", "sessions", len(exported), "bytes", len(exportData))
+ return sessions, nil
+}
+
+// decodeBackupPrivateKey extracts the raw 32-byte Curve25519 private key.
+// Tries several base64 encodings used in the wild before giving up.
+func decodeBackupPrivateKey(raw []byte) ([]byte, error) {
+ if len(raw) == 32 {
+ return raw, nil
+ }
+ s := strings.TrimSpace(string(raw))
+ for _, enc := range []*base64.Encoding{
+ base64.RawStdEncoding,
+ base64.StdEncoding,
+ base64.RawURLEncoding,
+ base64.URLEncoding,
+ } {
+ if b, err := enc.DecodeString(s); err == nil && len(b) == 32 {
+ return b, nil
+ }
+ }
+ return nil, fmt.Errorf("cannot decode %d-byte value as a 32-byte Curve25519 private key", len(raw))
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Event decryption
+// ─────────────────────────────────────────────────────────────────────────────
+
+// tryDecryptEvent attempts in-place Megolm decryption of an m.room.encrypted
+// event. Returns the event unchanged if decryption is not possible.
+func tryDecryptEvent(ev *event.Event, sessions megolmSessions) *event.Event {
+ if ev.Type != event.EventEncrypted || sessions == nil {
+ return ev
+ }
+ // Events from client.Messages() only have VeryRaw set; ParseRaw must be
+ // called before the typed helpers (AsEncrypted etc.) work.
+ if ev.Content.Parsed == nil {
+ _ = ev.Content.ParseRaw(ev.Type)
+ }
+ content := ev.Content.AsEncrypted()
+ if content.Algorithm != id.AlgorithmMegolmV1 {
+ return ev
+ }
+ sess, ok := sessions[content.SessionID]
+ if !ok {
+ return ev
+ }
+ plaintext, _, err := sess.Decrypt(content.MegolmCiphertext)
+ if err != nil {
+ slog.Debug("Megolm decrypt failed", "session_id", content.SessionID, "error", err)
+ return ev
+ }
+ var inner struct {
+ Type string `json:"type"`
+ Content json.RawMessage `json:"content"`
+ }
+ if err := json.Unmarshal(plaintext, &inner); err != nil {
+ return ev
+ }
+ decrypted := *ev
+ decrypted.Type = event.NewEventType(inner.Type)
+ decrypted.Content = event.Content{VeryRaw: inner.Content}
+ return &decrypted
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Megolm key export (spec §14.4 — PBKDF2-SHA512 + AES-256-CTR + HMAC-SHA256)
+// ─────────────────────────────────────────────────────────────────────────────
+
+func buildMegolmExport(sessions []exportedSessionEntry, passphrase string) ([]byte, error) {
+ salt := make([]byte, 16)
+ if _, err := rand.Read(salt); err != nil {
+ return nil, err
+ }
+ ivBytes := make([]byte, 16)
+ if _, err := rand.Read(ivBytes); err != nil {
+ return nil, err
+ }
+ ivBytes[0] &= 0x7F // spec: highest bit of IV must be 0
+
+ const iterations = 100_000
+ keyMaterial := pbkdf2.Key([]byte(passphrase), salt, iterations, 64, sha512.New)
+ aesKey := keyMaterial[:32]
+ hmacKey := keyMaterial[32:]
+
+ plaintext, err := json.Marshal(sessions)
+ if err != nil {
+ return nil, err
+ }
+ block, err := aes.NewCipher(aesKey)
+ if err != nil {
+ return nil, err
+ }
+ ciphertext := make([]byte, len(plaintext))
+ cipher.NewCTR(block, ivBytes).XORKeyStream(ciphertext, plaintext)
+
+ iterBuf := make([]byte, 4)
+ binary.BigEndian.PutUint32(iterBuf, uint32(iterations))
+
+ payload := []byte{0x01}
+ payload = append(payload, salt...)
+ payload = append(payload, ivBytes...)
+ payload = append(payload, iterBuf...)
+ payload = append(payload, ciphertext...)
+
+ mac := hmac.New(sha256.New, hmacKey)
+ mac.Write(payload)
+ payload = append(payload, mac.Sum(nil)...)
+
+ encoded := base64.StdEncoding.EncodeToString(payload)
+ var sb strings.Builder
+ sb.WriteString("-----BEGIN MEGOLM SESSION DATA-----\n")
+ for i := 0; i < len(encoded); i += 76 {
+ end := i + 76
+ if end > len(encoded) {
+ end = len(encoded)
+ }
+ sb.WriteString(encoded[i:end])
+ sb.WriteByte('\n')
+ }
+ sb.WriteString("-----END MEGOLM SESSION DATA-----\n")
+ return []byte(sb.String()), nil
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/main.go b/apps/talos_cluster/matrix-backup/backup-tool/main.go
@@ -11,1754 +11,11 @@
package main
import (
- "archive/tar"
- "bytes"
- "compress/gzip"
"context"
- "crypto/aes"
- "crypto/cipher"
- "crypto/hmac"
- "crypto/rand"
- "crypto/sha256"
- "crypto/sha512"
- "encoding/base64"
- "encoding/binary"
- "encoding/json"
- "errors"
- "fmt"
- "io"
"log/slog"
- "net/http"
- "net/url"
"os"
- "strings"
- "time"
-
- "filippo.io/age"
- "github.com/aws/aws-sdk-go-v2/aws"
- awsconfig "github.com/aws/aws-sdk-go-v2/config"
- "github.com/aws/aws-sdk-go-v2/credentials"
- "github.com/aws/aws-sdk-go-v2/service/s3"
- s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
- "golang.org/x/crypto/pbkdf2"
- "maunium.net/go/mautrix"
- "maunium.net/go/mautrix/crypto/backup"
- "maunium.net/go/mautrix/crypto/goolm/session"
- "maunium.net/go/mautrix/crypto/ssss"
- "maunium.net/go/mautrix/event"
- "maunium.net/go/mautrix/id"
-)
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Config
-// ─────────────────────────────────────────────────────────────────────────────
-
-const ourServerName = "mtrnord.blog"
-
-type accountCfg struct {
- UserID id.UserID
- Password string
- SSSSKey string
- Prefix string
- StoreDir string
-}
-
-var (
- homeserver = mustEnv("HOMESERVER")
- s3Endpoint = mustEnv("S3_ENDPOINT")
- s3BucketName = mustEnv("S3_BUCKET")
- s3AccessKey = mustEnv("S3_ACCESS_KEY")
- s3SecretKey = mustEnv("S3_SECRET_KEY")
- keyExportPass = mustEnv("KEY_EXPORT_PASSPHRASE")
- dateStr = time.Now().UTC().Format("20060102")
- runStr = time.Now().UTC().Format("20060102-150405")
-
- accounts = []accountCfg{
- {
- UserID: "@mtrnord:mtrnord.blog",
- Password: os.Getenv("MTRNORD_PASSWORD"),
- SSSSKey: os.Getenv("MTRNORD_SSSS_KEY"),
- Prefix: "mtrnord",
- StoreDir: "/data/crypto/mtrnord",
- },
- {
- UserID: "@lexi:mtrnord.blog",
- Password: os.Getenv("LEXI_PASSWORD"),
- SSSSKey: os.Getenv("LEXI_SSSS_KEY"),
- Prefix: "lexi",
- StoreDir: "/data/crypto/lexi",
- },
- }
-
- // ageRecipients holds the parsed public keys used to encrypt history files.
- ageRecipients []age.Recipient
-
- // ageIdentity is the optional private key used to decrypt files written by
- // this tool (loaded from AGE_PRIVATE_KEY). Allows reading back previously
- // encrypted objects (e.g. the room list) without storing plain copies in S3.
- ageIdentity age.Identity
)
-func mustEnv(key string) string {
- v := os.Getenv(key)
- if v == "" {
- slog.Error("Required environment variable not set", "key", key)
- os.Exit(1)
- }
- return v
-}
-
-// initAgeRecipients parses a comma-separated list of age public keys and
-// populates the global ageRecipients slice.
-func initAgeRecipients(s string) error {
- for _, raw := range strings.Split(s, ",") {
- raw = strings.TrimSpace(raw)
- if raw == "" {
- continue
- }
- r, err := age.ParseX25519Recipient(raw)
- if err != nil {
- return fmt.Errorf("invalid age recipient %q: %w", raw, err)
- }
- ageRecipients = append(ageRecipients, r)
- }
- if len(ageRecipients) == 0 {
- return fmt.Errorf("AGE_RECIPIENTS is empty — at least one public key is required")
- }
- return nil
-}
-
-// initAgeIdentity parses the AGE_PRIVATE_KEY value (a single AGE-SECRET-KEY-1…
-// line) and stores it for use when decrypting previously written .age files.
-func initAgeIdentity(privKey string) error {
- privKey = strings.TrimSpace(privKey)
- if privKey == "" {
- return nil
- }
- ids, err := age.ParseIdentities(strings.NewReader(privKey))
- if err != nil {
- return fmt.Errorf("parse age identity: %w", err)
- }
- if len(ids) > 0 {
- ageIdentity = ids[0]
- }
- return nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Age encryption / decryption
-// ─────────────────────────────────────────────────────────────────────────────
-
-// ageEncrypt encrypts data for all ageRecipients and returns the ciphertext.
-func ageEncrypt(data []byte) ([]byte, error) {
- var buf bytes.Buffer
- w, err := age.Encrypt(&buf, ageRecipients...)
- if err != nil {
- return nil, err
- }
- if _, err := w.Write(data); err != nil {
- return nil, err
- }
- if err := w.Close(); err != nil {
- return nil, err
- }
- return buf.Bytes(), nil
-}
-
-// getDecryptedAgeFromS3 fetches an .age object from S3 and decrypts it using
-// the configured ageIdentity. Returns (nil, nil) when the object doesn't exist
-// or no identity is available.
-func getDecryptedAgeFromS3(ctx context.Context, key string) ([]byte, error) {
- if ageIdentity == nil {
- return nil, nil
- }
- enc, err := s3Get(ctx, key)
- if err != nil || enc == nil {
- return nil, err
- }
- r, err := age.Decrypt(bytes.NewReader(enc), ageIdentity)
- if err != nil {
- return nil, fmt.Errorf("age decrypt %s: %w", key, err)
- }
- return io.ReadAll(r)
-}
-
-// s3PutAge age-encrypts data then uploads it under key (appending ".age").
-func s3PutAge(ctx context.Context, key string, data []byte) error {
- enc, err := ageEncrypt(data)
- if err != nil {
- return fmt.Errorf("age encrypt %s: %w", key, err)
- }
- return s3Put(ctx, key+".age", enc, "application/octet-stream")
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// S3 helpers
-// ─────────────────────────────────────────────────────────────────────────────
-
-var s3c *s3.Client
-
-func initS3(ctx context.Context) error {
- cfg, err := awsconfig.LoadDefaultConfig(ctx,
- awsconfig.WithRegion("hel1"),
- awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
- s3AccessKey, s3SecretKey, "",
- )),
- )
- if err != nil {
- return err
- }
- endpoint := "https://" + s3Endpoint
- s3c = s3.NewFromConfig(cfg, func(o *s3.Options) {
- o.UsePathStyle = true
- o.BaseEndpoint = &endpoint
- })
- return nil
-}
-
-func s3Get(ctx context.Context, key string) ([]byte, error) {
- out, err := s3c.GetObject(ctx, &s3.GetObjectInput{
- Bucket: aws.String(s3BucketName),
- Key: aws.String(key),
- })
- if err != nil {
- var nsk *s3types.NoSuchKey
- if errors.As(err, &nsk) {
- return nil, nil
- }
- return nil, err
- }
- defer out.Body.Close()
- return io.ReadAll(out.Body)
-}
-
-func s3GetJSON(ctx context.Context, key string, out interface{}) error {
- data, err := s3Get(ctx, key)
- if err != nil || data == nil {
- return err
- }
- return json.Unmarshal(data, out)
-}
-
-func s3Put(ctx context.Context, key string, data []byte, contentType string) error {
- _, err := s3c.PutObject(ctx, &s3.PutObjectInput{
- Bucket: aws.String(s3BucketName),
- Key: aws.String(key),
- Body: bytes.NewReader(data),
- ContentType: aws.String(contentType),
- })
- return err
-}
-
-func s3Exists(ctx context.Context, key string) bool {
- _, err := s3c.HeadObject(ctx, &s3.HeadObjectInput{
- Bucket: aws.String(s3BucketName),
- Key: aws.String(key),
- })
- return err == nil
-}
-
-// s3DeletePrefix deletes all objects whose key starts with prefix.
-func s3DeletePrefix(ctx context.Context, prefix string) error {
- paginator := s3.NewListObjectsV2Paginator(s3c, &s3.ListObjectsV2Input{
- Bucket: aws.String(s3BucketName),
- Prefix: aws.String(prefix),
- })
- deleted := 0
- for paginator.HasMorePages() {
- page, err := paginator.NextPage(ctx)
- if err != nil {
- return err
- }
- for _, obj := range page.Contents {
- if _, err := s3c.DeleteObject(ctx, &s3.DeleteObjectInput{
- Bucket: aws.String(s3BucketName),
- Key: obj.Key,
- }); err != nil {
- slog.Warn("Failed to delete S3 object", "key", *obj.Key, "error", err)
- } else {
- deleted++
- }
- }
- }
- slog.Info("Deleted S3 prefix", "prefix", prefix, "count", deleted)
- return nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Crypto-store tarball (persists sessions and sync tokens across CronJob runs)
-// ─────────────────────────────────────────────────────────────────────────────
-
-// downloadStore restores a previously uploaded crypto store tarball from S3.
-// Missing store (first run) is not an error — the caller starts fresh.
-func downloadStore(ctx context.Context, storeDir, s3Key string) error {
- data, err := s3Get(ctx, s3Key)
- if err != nil {
- return err
- }
- if data == nil {
- slog.Info("No existing store in S3, starting fresh", "key", s3Key)
- return nil
- }
- if err := os.MkdirAll(storeDir, 0700); err != nil {
- return err
- }
- gr, err := gzip.NewReader(bytes.NewReader(data))
- if err != nil {
- return err
- }
- defer gr.Close()
- tr := tar.NewReader(gr)
- for {
- hdr, err := tr.Next()
- if errors.Is(err, io.EOF) {
- break
- }
- if err != nil {
- return err
- }
- target := storeDir + "/" + hdr.Name
- switch hdr.Typeflag {
- case tar.TypeDir:
- os.MkdirAll(target, 0700) //nolint:errcheck
- case tar.TypeReg:
- f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
- if err != nil {
- return err
- }
- if _, err := io.Copy(f, tr); err != nil {
- f.Close()
- return err
- }
- f.Close()
- }
- }
- slog.Info("Store restored from S3", "key", s3Key, "bytes", len(data))
- return nil
-}
-
-// uploadStore tarballs the crypto store directory and uploads it to S3 for the
-// next run to restore.
-func uploadStore(ctx context.Context, storeDir, s3Key string) error {
- var buf bytes.Buffer
- gw := gzip.NewWriter(&buf)
- tw := tar.NewWriter(gw)
- if err := addDirToTar(tw, storeDir, "."); err != nil {
- return err
- }
- tw.Close() //nolint:errcheck
- gw.Close() //nolint:errcheck
- data := buf.Bytes()
- if err := s3Put(ctx, s3Key, data, "application/gzip"); err != nil {
- return err
- }
- slog.Info("Store saved to S3", "key", s3Key, "bytes", len(data))
- return nil
-}
-
-func addDirToTar(tw *tar.Writer, baseDir, arcBase string) error {
- entries, err := os.ReadDir(baseDir)
- if err != nil {
- if os.IsNotExist(err) {
- return nil
- }
- return err
- }
- for _, e := range entries {
- srcPath := baseDir + "/" + e.Name()
- arcPath := arcBase + "/" + e.Name()
- if e.IsDir() {
- _ = tw.WriteHeader(&tar.Header{
- Typeflag: tar.TypeDir,
- Name: arcPath + "/",
- Mode: 0700,
- })
- if err := addDirToTar(tw, srcPath, arcPath); err != nil {
- return err
- }
- continue
- }
- info, err := e.Info()
- if err != nil {
- continue
- }
- f, err := os.Open(srcPath)
- if err != nil {
- continue
- }
- if err := tw.WriteHeader(&tar.Header{
- Typeflag: tar.TypeReg,
- Name: arcPath,
- Size: info.Size(),
- Mode: 0600,
- }); err != nil {
- f.Close()
- return err
- }
- _, copyErr := io.Copy(tw, f)
- f.Close()
- if copyErr != nil {
- return copyErr
- }
- }
- return nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Session management
-// ─────────────────────────────────────────────────────────────────────────────
-
-type sessionData struct {
- AccessToken string `json:"access_token"`
- DeviceID id.DeviceID `json:"device_id"`
-}
-
-// ensureSession loads a saved session from S3 or logs in with the account
-// password and saves the resulting session for future runs.
-func ensureSession(ctx context.Context, client *mautrix.Client, acc accountCfg, s3Key string) (*sessionData, error) {
- var sess sessionData
- if err := s3GetJSON(ctx, s3Key, &sess); err != nil {
- return nil, fmt.Errorf("load session: %w", err)
- }
- if sess.AccessToken != "" {
- slog.Info("Restored existing session", "device_id", sess.DeviceID)
- client.UserID = acc.UserID
- client.DeviceID = sess.DeviceID
- client.AccessToken = sess.AccessToken
- return &sess, nil
- }
-
- slog.Info("First run — logging in with password", "user_id", acc.UserID)
- resp, err := client.Login(ctx, &mautrix.ReqLogin{
- Type: mautrix.AuthTypePassword,
- Identifier: mautrix.UserIdentifier{
- Type: mautrix.IdentifierTypeUser,
- User: string(acc.UserID),
- },
- Password: acc.Password,
- InitialDeviceDisplayName: "matrix-backup",
- })
- if err != nil {
- return nil, fmt.Errorf("login: %w", err)
- }
- sess = sessionData{
- AccessToken: resp.AccessToken,
- DeviceID: resp.DeviceID,
- }
- client.UserID = resp.UserID
- client.DeviceID = resp.DeviceID
- client.AccessToken = resp.AccessToken
- slog.Info("Logged in", "device_id", resp.DeviceID)
-
- data, err := json.Marshal(sess)
- if err != nil {
- return nil, err
- }
- if err := s3Put(ctx, s3Key, data, "application/json"); err != nil {
- return nil, fmt.Errorf("save session: %w", err)
- }
- return &sess, nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Matrix API helpers
-// ─────────────────────────────────────────────────────────────────────────────
-
-// matrixGetJSON makes an authenticated GET to the homeserver and JSON-decodes
-// the response. Used for endpoints not wrapped by the mautrix client.
-func matrixGetJSON(ctx context.Context, client *mautrix.Client, path string, out interface{}) error {
- base := strings.TrimRight(client.HomeserverURL.String(), "/")
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+path, nil)
- if err != nil {
- return err
- }
- req.Header.Set("Authorization", "Bearer "+client.AccessToken)
- resp, err := http.DefaultClient.Do(req)
- if err != nil {
- return err
- }
- defer resp.Body.Close()
- body, err := io.ReadAll(resp.Body)
- if err != nil {
- return err
- }
- if resp.StatusCode != http.StatusOK {
- return fmt.Errorf("HTTP %d: %s", resp.StatusCode, body)
- }
- return json.Unmarshal(body, out)
-}
-
-func getAccountData(ctx context.Context, client *mautrix.Client, eventType string, out interface{}) error {
- path := "/_matrix/client/v3/user/" +
- url.PathEscape(client.UserID.String()) +
- "/account_data/" +
- url.PathEscape(eventType)
- return matrixGetJSON(ctx, client, path, out)
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// SSSS + Megolm key backup
-// ─────────────────────────────────────────────────────────────────────────────
-
-type defaultKeyEventContent struct {
- Key string `json:"key"`
-}
-
-// encryptedSecretContent matches the `{"encrypted": {"<keyID>": {...}}}` format.
-type encryptedSecretContent struct {
- Encrypted map[string]ssss.EncryptedKeyData `json:"encrypted"`
-}
-
-type keyBackupVersionResp struct {
- Version string `json:"version"`
- Algorithm string `json:"algorithm"`
-}
-
-type keyBackupRoomSession struct {
- FirstMessageIndex int `json:"first_message_index"`
- ForwardedCount int `json:"forwarded_count"`
- IsVerified bool `json:"is_verified"`
- SessionData json.RawMessage `json:"session_data"`
-}
-
-type keyBackupRoom struct {
- Sessions map[string]keyBackupRoomSession `json:"sessions"`
-}
-
-type keyBackupAllRooms struct {
- Rooms map[id.RoomID]keyBackupRoom `json:"rooms"`
-}
-
-// exportedSessionEntry matches the standard Megolm key export session format
-// (https://spec.matrix.org/v1.9/client-server-api/#key-exports).
-type exportedSessionEntry struct {
- Algorithm string `json:"algorithm"`
- ForwardingCurve25519KeyChain []string `json:"forwarding_curve25519_key_chain"`
- RoomID string `json:"room_id"`
- SenderClaimedKeys map[string]string `json:"sender_claimed_keys"`
- SenderKey string `json:"sender_key"`
- SessionID string `json:"session_id"`
- SessionKey string `json:"session_key"`
-}
-
-// megolmSessions maps session IDs to live inbound sessions ready to decrypt.
-type megolmSessions map[id.SessionID]*session.MegolmInboundSession
-
-// fetchAndExportKeyBackup derives the Megolm backup private key from SSSS,
-// decrypts every backed-up session, and uploads both a raw JSON archive and a
-// standard .key export file (importable by any Matrix client) to S3.
-// It also returns an in-memory session map for use when decrypting room history.
-func fetchAndExportKeyBackup(ctx context.Context, client *mautrix.Client, recoveryKeyStr, prefix string) (megolmSessions, error) {
- var defaultKey defaultKeyEventContent
- if err := getAccountData(ctx, client, "m.secret_storage.default_key", &defaultKey); err != nil {
- return nil, fmt.Errorf("get default key: %w", err)
- }
- if defaultKey.Key == "" {
- return nil, fmt.Errorf("m.secret_storage.default_key has no 'key' field")
- }
- keyID := defaultKey.Key
- slog.Info("SSSS default key ID", "key_id", keyID)
-
- var keyMetadata ssss.KeyMetadata
- if err := getAccountData(ctx, client, "m.secret_storage.key."+keyID, &keyMetadata); err != nil {
- return nil, fmt.Errorf("get key metadata: %w", err)
- }
-
- // ErrUnverifiableKey just means the metadata has no MAC to check against —
- // the derived key is still usable.
- sssKey, err := keyMetadata.VerifyRecoveryKey(keyID, recoveryKeyStr)
- if err != nil && !errors.Is(err, ssss.ErrUnverifiableKey) {
- return nil, fmt.Errorf("verify recovery key: %w", err)
- }
- slog.Info("Recovery key verified (or unverifiable but accepted)")
-
- var backupSecret encryptedSecretContent
- if err := getAccountData(ctx, client, "m.megolm_backup.v1", &backupSecret); err != nil {
- return nil, fmt.Errorf("get backup secret: %w", err)
- }
- encData, ok := backupSecret.Encrypted[keyID]
- if !ok {
- return nil, fmt.Errorf("backup secret not encrypted with key %q", keyID)
- }
- backupKeyRaw, err := sssKey.Decrypt("m.megolm_backup.v1", encData)
- if err != nil {
- return nil, fmt.Errorf("decrypt backup key from SSSS: %w", err)
- }
-
- privateKeyBytes, err := decodeBackupPrivateKey(backupKeyRaw)
- if err != nil {
- return nil, fmt.Errorf("decode backup private key: %w", err)
- }
- megolmKey, err := backup.MegolmBackupKeyFromBytes(privateKeyBytes)
- if err != nil {
- return nil, fmt.Errorf("create megolm backup key: %w", err)
- }
- slog.Info("Decoded megolm backup private key from SSSS")
-
- var backupVersion keyBackupVersionResp
- if err := matrixGetJSON(ctx, client, "/_matrix/client/v3/room_keys/version", &backupVersion); err != nil {
- return nil, fmt.Errorf("get backup version: %w", err)
- }
- slog.Info("Key backup version", "version", backupVersion.Version)
-
- var allRooms keyBackupAllRooms
- if err := matrixGetJSON(ctx, client,
- "/_matrix/client/v3/room_keys/keys?version="+url.QueryEscape(backupVersion.Version),
- &allRooms,
- ); err != nil {
- return nil, fmt.Errorf("get backup keys: %w", err)
- }
- slog.Info("Fetched key backup", "rooms", len(allRooms.Rooms))
-
- // Decrypt each session: build both the export list and the live session map.
- sessions := make(megolmSessions)
- var exported []exportedSessionEntry
- decOK, decFail := 0, 0
- for roomID, room := range allRooms.Rooms {
- for sessionID, sessionInfo := range room.Sessions {
- var encSD backup.EncryptedSessionData[backup.MegolmSessionData]
- if err := json.Unmarshal(sessionInfo.SessionData, &encSD); err != nil {
- slog.Warn("Failed to parse session_data",
- "room_id", roomID, "session_id", sessionID, "error", err)
- decFail++
- continue
- }
- sessionData, err := encSD.Decrypt(megolmKey)
- if err != nil {
- slog.Warn("Failed to decrypt session",
- "room_id", roomID, "session_id", sessionID, "error", err)
- decFail++
- continue
- }
-
- // Build the in-memory session for history decryption.
- if sess, err := session.NewMegolmInboundSessionFromExport([]byte(sessionData.SessionKey)); err == nil {
- sessions[id.SessionID(sessionID)] = sess
- } else {
- slog.Warn("Failed to import Megolm session", "session_id", sessionID, "error", err)
- }
-
- exported = append(exported, exportedSessionEntry{
- Algorithm: string(sessionData.Algorithm),
- ForwardingCurve25519KeyChain: sessionData.ForwardingKeyChain,
- RoomID: string(roomID),
- SenderClaimedKeys: map[string]string{"ed25519": string(sessionData.SenderClaimedKeys.Ed25519)},
- SenderKey: string(sessionData.SenderKey),
- SessionID: sessionID,
- SessionKey: sessionData.SessionKey,
- })
- decOK++
- }
- }
- slog.Info("Session decryption complete", "ok", decOK, "failed", decFail, "in_memory", len(sessions))
-
- if len(exported) == 0 {
- slog.Warn("No sessions decrypted — skipping key export")
- return sessions, nil
- }
-
- if rawJSON, err := json.MarshalIndent(exported, "", " "); err == nil {
- _ = s3PutAge(ctx, prefix+"/backup-sessions-latest.json", rawJSON)
- _ = s3PutAge(ctx, prefix+"/backup-sessions-"+dateStr+".json", rawJSON)
- }
-
- exportData, err := buildMegolmExport(exported, keyExportPass)
- if err != nil {
- return sessions, fmt.Errorf("build megolm export: %w", err)
- }
- if err := s3PutAge(ctx, prefix+"/crypto-keys-latest.bin", exportData); err != nil {
- return sessions, err
- }
- if err := s3PutAge(ctx, prefix+"/crypto-keys-"+dateStr+".bin", exportData); err != nil {
- return sessions, err
- }
- slog.Info("Exported E2EE keys", "sessions", len(exported), "bytes", len(exportData))
- return sessions, nil
-}
-
-// decodeBackupPrivateKey extracts the raw 32-byte Curve25519 private key from
-// whatever format the client stored it as. The spec says unpadded base64, but
-// in practice standard, URL-safe, and padded variants all appear in the wild.
-// If the bytes are already 32 bytes long they're used directly.
-func decodeBackupPrivateKey(raw []byte) ([]byte, error) {
- if len(raw) == 32 {
- return raw, nil
- }
- s := strings.TrimSpace(string(raw))
- for _, enc := range []*base64.Encoding{
- base64.RawStdEncoding,
- base64.StdEncoding,
- base64.RawURLEncoding,
- base64.URLEncoding,
- } {
- if b, err := enc.DecodeString(s); err == nil && len(b) == 32 {
- return b, nil
- }
- }
- return nil, fmt.Errorf("cannot decode %d-byte value as a 32-byte Curve25519 private key", len(raw))
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Megolm event decryption
-// ─────────────────────────────────────────────────────────────────────────────
-
-// tryDecryptEvent attempts to decrypt an m.room.encrypted event using the
-// in-memory session map. Returns the original event unchanged if decryption
-// isn't possible (missing session, wrong algorithm, parse error).
-func tryDecryptEvent(ev *event.Event, sessions megolmSessions) *event.Event {
- if ev.Type != event.EventEncrypted || sessions == nil {
- return ev
- }
- // Events from client.Messages() only have VeryRaw set; Parsed is nil until
- // ParseRaw is called explicitly. AsEncrypted() returns an empty struct (not
- // nil) when Parsed is absent, so Algorithm would be "" and we'd bail early.
- if ev.Content.Parsed == nil {
- _ = ev.Content.ParseRaw(ev.Type)
- }
- content := ev.Content.AsEncrypted()
- if content.Algorithm != id.AlgorithmMegolmV1 {
- return ev
- }
- sess, ok := sessions[content.SessionID]
- if !ok {
- return ev
- }
- plaintext, _, err := sess.Decrypt(content.MegolmCiphertext)
- if err != nil {
- slog.Debug("Megolm decrypt failed", "session_id", content.SessionID, "error", err)
- return ev
- }
- var inner struct {
- Type string `json:"type"`
- Content json.RawMessage `json:"content"`
- }
- if err := json.Unmarshal(plaintext, &inner); err != nil {
- return ev
- }
- decrypted := *ev
- decrypted.Type = event.NewEventType(inner.Type)
- decrypted.Content = event.Content{VeryRaw: inner.Content}
- return &decrypted
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Megolm key export (spec: https://spec.matrix.org/v1.9/client-server-api/#key-exports)
-// ─────────────────────────────────────────────────────────────────────────────
-
-// buildMegolmExport encodes sessions into the standard passphrase-protected key
-// export format used by all Matrix clients (PBKDF2-SHA512 + AES-256-CTR + HMAC-SHA256).
-func buildMegolmExport(sessions []exportedSessionEntry, passphrase string) ([]byte, error) {
- salt := make([]byte, 16)
- if _, err := rand.Read(salt); err != nil {
- return nil, err
- }
- ivBytes := make([]byte, 16)
- if _, err := rand.Read(ivBytes); err != nil {
- return nil, err
- }
- ivBytes[0] &= 0x7F // spec: highest bit of IV must be 0
-
- const iterations = 100_000
- keyMaterial := pbkdf2.Key([]byte(passphrase), salt, iterations, 64, sha512.New)
- aesKey := keyMaterial[:32]
- hmacKey := keyMaterial[32:]
-
- plaintext, err := json.Marshal(sessions)
- if err != nil {
- return nil, err
- }
-
- block, err := aes.NewCipher(aesKey)
- if err != nil {
- return nil, err
- }
- ciphertext := make([]byte, len(plaintext))
- cipher.NewCTR(block, ivBytes).XORKeyStream(ciphertext, plaintext)
-
- iterBuf := make([]byte, 4)
- binary.BigEndian.PutUint32(iterBuf, uint32(iterations))
-
- payload := []byte{0x01}
- payload = append(payload, salt...)
- payload = append(payload, ivBytes...)
- payload = append(payload, iterBuf...)
- payload = append(payload, ciphertext...)
-
- mac := hmac.New(sha256.New, hmacKey)
- mac.Write(payload)
- payload = append(payload, mac.Sum(nil)...)
-
- encoded := base64.StdEncoding.EncodeToString(payload)
- var sb strings.Builder
- sb.WriteString("-----BEGIN MEGOLM SESSION DATA-----\n")
- for i := 0; i < len(encoded); i += 76 {
- end := i + 76
- if end > len(encoded) {
- end = len(encoded)
- }
- sb.WriteString(encoded[i:end])
- sb.WriteByte('\n')
- }
- sb.WriteString("-----END MEGOLM SESSION DATA-----\n")
- return []byte(sb.String()), nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Room list
-// ─────────────────────────────────────────────────────────────────────────────
-
-// calcRoomName implements the Matrix room display name algorithm from the spec:
-// https://spec.matrix.org/v1.11/client-server-api/#calculating-the-display-name-for-a-room
-//
-// Priority: explicit m.room.name → canonical alias → heroes list → "Empty Room".
-func calcRoomName(
- stateEvents []*event.Event,
- heroes []id.UserID,
- joinedCount, invitedCount int,
- selfUserID id.UserID,
-) string {
- // 1. Explicit room name.
- for _, ev := range stateEvents {
- if ev.Type == event.StateRoomName {
- if c := ev.Content.AsRoomName(); c != nil && c.Name != "" {
- return c.Name
- }
- }
- }
-
- // 2. Canonical alias.
- for _, ev := range stateEvents {
- if ev.Type == event.StateCanonicalAlias {
- if c := ev.Content.AsCanonicalAlias(); c != nil && c.Alias != "" {
- return string(c.Alias)
- }
- }
- }
-
- // 3. Heroes list — build display names from member state events.
- memberNames := make(map[id.UserID]string)
- for _, ev := range stateEvents {
- if ev.Type == event.StateMember && ev.StateKey != nil {
- uid := id.UserID(*ev.StateKey)
- if c := ev.Content.AsMember(); c != nil && c.Displayname != "" {
- memberNames[uid] = c.Displayname
- } else {
- local, _, _ := uid.Parse()
- if local != "" {
- memberNames[uid] = local
- } else {
- memberNames[uid] = string(uid)
- }
- }
- }
- }
-
- // Spec says heroes should already exclude self, but filter just in case.
- var filtered []id.UserID
- for _, h := range heroes {
- if h != selfUserID {
- filtered = append(filtered, h)
- }
- }
-
- // Build tentative names for each hero, then disambiguate duplicates by
- // appending the server part (e.g. "mtrnord (mtrnord.blog)").
- tentativeNames := make(map[id.UserID]string, len(filtered))
- for _, h := range filtered {
- if n, ok := memberNames[h]; ok {
- tentativeNames[h] = n
- } else {
- local, _, _ := h.Parse()
- if local != "" {
- tentativeNames[h] = local
- } else {
- tentativeNames[h] = string(h)
- }
- }
- }
- nameCounts := make(map[string]int, len(filtered))
- for _, n := range tentativeNames {
- nameCounts[n]++
- }
- heroName := func(uid id.UserID) string {
- n := tentativeNames[uid]
- if nameCounts[n] > 1 {
- _, server, err := uid.Parse()
- if err == nil && server != "" {
- return n + " (" + server + ")"
- }
- return string(uid)
- }
- return n
- }
-
- // Total other members in the room (everyone except self).
- others := joinedCount + invitedCount - 1
- if others < 0 {
- others = 0
- }
-
- if len(filtered) == 0 {
- if others == 0 {
- return "Empty Room"
- }
- // Heroes list absent but we know there are members — fall through to ID.
- return ""
- }
-
- names := make([]string, len(filtered))
- for i, h := range filtered {
- names[i] = heroName(h)
- }
-
- // How many members aren't represented by the heroes list.
- unnamed := others - len(names)
-
- if unnamed <= 0 {
- // All others are named.
- switch len(names) {
- case 1:
- return names[0]
- case 2:
- return names[0] + " and " + names[1]
- default:
- return strings.Join(names[:len(names)-1], ", ") + ", and " + names[len(names)-1]
- }
- }
-
- // Some members aren't in the heroes list.
- switch len(names) {
- case 1:
- return fmt.Sprintf("%s and %d others", names[0], unnamed)
- case 2:
- return fmt.Sprintf("%s, %s, and %d others", names[0], names[1], unnamed)
- default:
- return fmt.Sprintf("%s, and %d others", strings.Join(names, ", "), unnamed)
- }
-}
-
-type roomEntry struct {
- RoomID string `json:"room_id"`
- Name string `json:"name"`
- Type string `json:"type"`
- Aliases []string `json:"aliases"`
- CanonicalAlias string `json:"canonical_alias,omitempty"`
- MemberCount int `json:"member_count"`
- Encrypted bool `json:"encrypted"`
- ViaServers []string `json:"via_servers,omitempty"`
-}
-
-// collectViaServers returns up to 3 server names suitable for matrix.to
-// ?via= parameters. Priority: room-ID server → heroes → joined members from
-// state events (heroes are absent from incremental sync summaries).
-func collectViaServers(roomID id.RoomID, heroes []id.UserID, stateEvents []*event.Event) []string {
- seen := make(map[string]bool)
- var servers []string
-
- add := func(s string) bool {
- if s != "" && !seen[s] {
- seen[s] = true
- servers = append(servers, s)
- }
- return len(servers) >= 3
- }
-
- // Server embedded in the room ID (old-style !localpart:server).
- if idx := strings.LastIndex(string(roomID), ":"); idx >= 0 {
- add(string(roomID)[idx+1:])
- }
- // Heroes from sync summary.
- for _, h := range heroes {
- _, server, err := h.Parse()
- if err == nil && add(server) {
- return servers
- }
- }
- // Joined members from state events (present even when heroes summary is empty).
- for _, ev := range stateEvents {
- if len(servers) >= 3 {
- break
- }
- if ev.Type != event.StateMember || ev.StateKey == nil {
- continue
- }
- var mc struct {
- Membership string `json:"membership"`
- }
- if ev.Content.VeryRaw != nil {
- _ = json.Unmarshal(ev.Content.VeryRaw, &mc)
- }
- if mc.Membership != "join" {
- continue
- }
- _, server, err := id.UserID(*ev.StateKey).Parse()
- if err == nil {
- add(server)
- }
- }
- return servers
-}
-
-// fetchRoomNameFromState fetches m.room.name then m.room.canonical_alias
-// directly from the server state API. Used when the sync delta doesn't include
-// those events (common for large/public rooms that haven't changed recently).
-func fetchRoomNameFromState(ctx context.Context, client *mautrix.Client, roomID id.RoomID) string {
- base := url.PathEscape(string(roomID))
-
- var nameContent struct {
- Name string `json:"name"`
- }
- if err := matrixGetJSON(ctx, client,
- "/_matrix/client/v3/rooms/"+base+"/state/m.room.name",
- &nameContent,
- ); err == nil && nameContent.Name != "" {
- return nameContent.Name
- }
-
- var aliasContent struct {
- Alias string `json:"alias"`
- }
- if err := matrixGetJSON(ctx, client,
- "/_matrix/client/v3/rooms/"+base+"/state/m.room.canonical_alias",
- &aliasContent,
- ); err == nil && aliasContent.Alias != "" {
- return aliasContent.Alias
- }
-
- return ""
-}
-
-// saveRoomList builds a JSON snapshot of all joined rooms (name, type, aliases,
-// member count, encryption status) and uploads it to S3.
-//
-// On incremental syncs the server only returns rooms with recent activity, so
-// we load the previously saved list and merge: rooms present in this sync get
-// fresh data, rooms absent from this sync are kept as-is from the prior list.
-func saveRoomList(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync, prefix string) error {
- // Load the previously persisted room list so that rooms absent from this
- // sync's delta are not lost.
- existing := make(map[string]roomEntry)
- if raw, err := getDecryptedAgeFromS3(ctx, prefix+"/rooms-latest.json.age"); err == nil && raw != nil {
- var prev []roomEntry
- if json.Unmarshal(raw, &prev) == nil {
- for _, r := range prev {
- // Retroactively reclassify 2-member non-space rooms as DMs.
- // This catches rooms stored as "normal" before the member-count
- // heuristic was added (e.g. DMs created by the other party).
- if r.Type == "normal" && r.MemberCount > 0 && r.MemberCount <= 2 {
- r.Type = "dm"
- }
- existing[r.RoomID] = r
- }
- }
- }
-
- dmRooms := getDMRooms(ctx, client, syncResp)
-
- for roomID, joinedRoom := range syncResp.Rooms.Join {
- var name, canonicalAlias string
- var aliases []string
- encrypted := false
- rtype := "normal"
- if dmRooms[roomID] {
- rtype = "dm"
- }
-
- // ParseRaw must be called explicitly — mautrix only sets VeryRaw during
- // JSON decode; the typed helpers (AsRoomName etc.) need Parsed != nil.
- for _, ev := range joinedRoom.State.Events {
- _ = ev.Content.ParseRaw(ev.Type)
- }
-
- for _, ev := range joinedRoom.State.Events {
- switch ev.Type {
- case event.StateRoomName:
- if c := ev.Content.AsRoomName(); c.Name != "" {
- name = c.Name
- }
- case event.StateCanonicalAlias:
- if c := ev.Content.AsCanonicalAlias(); c.Alias != "" {
- canonicalAlias = string(c.Alias)
- for _, a := range c.AltAliases {
- aliases = append(aliases, string(a))
- }
- }
- case event.StateEncryption:
- encrypted = true
- case event.StateCreate:
- if ev.Content.VeryRaw != nil {
- var createContent struct {
- Type string `json:"type"`
- }
- if json.Unmarshal(ev.Content.VeryRaw, &createContent) == nil && createContent.Type == "m.space" {
- rtype = "space"
- }
- }
- }
- }
-
- joinedCount, invitedCount := 0, 0
- if joinedRoom.Summary.JoinedMemberCount != nil {
- joinedCount = *joinedRoom.Summary.JoinedMemberCount
- }
- if joinedRoom.Summary.InvitedMemberCount != nil {
- invitedCount = *joinedRoom.Summary.InvitedMemberCount
- }
-
- // A 2-member non-space room is effectively a DM even if not in m.direct
- // (e.g. when the other party created the room and didn't populate our m.direct).
- if rtype == "normal" && joinedCount > 0 && joinedCount <= 2 {
- rtype = "dm"
- }
-
- if name == "" {
- name = calcRoomName(joinedRoom.State.Events, joinedRoom.Summary.Heroes, joinedCount, invitedCount, client.UserID)
- }
- if name == "" {
- // Sync delta doesn't include the name — fetch it directly.
- name = fetchRoomNameFromState(ctx, client, roomID)
- }
- // If we still have no name, keep whatever we had from the previous run.
- if name == "" {
- if prev, ok := existing[string(roomID)]; ok {
- name = prev.Name
- }
- }
- if name == "" {
- name = string(roomID)
- }
-
- existing[string(roomID)] = roomEntry{
- RoomID: string(roomID),
- Name: name,
- Type: rtype,
- Aliases: aliases,
- CanonicalAlias: canonicalAlias,
- MemberCount: joinedCount,
- Encrypted: encrypted,
- ViaServers: collectViaServers(roomID, joinedRoom.Summary.Heroes, joinedRoom.State.Events),
- }
- slog.Info("Room", "type", rtype, "name", name)
- }
-
- rooms := make([]roomEntry, 0, len(existing))
- for _, r := range existing {
- rooms = append(rooms, r)
- }
-
- data, err := json.MarshalIndent(rooms, "", " ")
- if err != nil {
- return err
- }
- // Age-encrypted copies — readable by both the backup tool (via ageIdentity)
- // and the external decrypt utility.
- if err := s3PutAge(ctx, prefix+"/rooms-"+dateStr+".json", data); err != nil {
- return err
- }
- if err := s3PutAge(ctx, prefix+"/rooms-latest.json", data); err != nil {
- return err
- }
- slog.Info("Uploaded room list", "rooms", len(rooms))
- return nil
-}
-
-// getDMRooms returns the full set of room IDs marked as direct chats.
-// It fetches m.direct via the account-data API so that incremental syncs
-// (which only carry changed account-data events) don't lose the DM map.
-func getDMRooms(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync) map[id.RoomID]bool {
- dmRooms := make(map[id.RoomID]bool)
-
- populate := func(direct event.DirectChatsEventContent) {
- for _, roomIDs := range direct {
- for _, rid := range roomIDs {
- dmRooms[rid] = true
- }
- }
- }
-
- // Always fetch from the API — the sync delta only carries m.direct when it
- // actually changes, so incremental runs would otherwise see an empty map.
- var direct event.DirectChatsEventContent
- if err := getAccountData(ctx, client, "m.direct", &direct); err == nil {
- populate(direct)
- return dmRooms
- }
-
- // Fallback: use whatever the sync included (first run or API error).
- for _, ev := range syncResp.AccountData.Events {
- if ev.Type == event.AccountDataDirectChats {
- if err := json.Unmarshal(ev.Content.VeryRaw, &direct); err == nil {
- populate(direct)
- }
- break
- }
- }
- return dmRooms
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Media download
-// ─────────────────────────────────────────────────────────────────────────────
-
-// downloadAndStoreMedia downloads a single mxc:// URL and stores it under
-// prefix/label/server/mediaID in S3, skipping if already present.
-func downloadAndStoreMedia(ctx context.Context, client *mautrix.Client, mxcURL, prefix, label string) error {
- if !strings.HasPrefix(mxcURL, "mxc://") {
- return nil
- }
- rest := mxcURL[len("mxc://"):]
- slashIdx := strings.Index(rest, "/")
- if slashIdx < 0 {
- return nil
- }
- server := rest[:slashIdx]
- mediaID := rest[slashIdx+1:]
- if server == "" || mediaID == "" {
- return nil
- }
-
- s3Key := prefix + "/" + label + "/" + server + "/" + mediaID
- if s3Exists(ctx, s3Key) {
- return nil
- }
-
- base := strings.TrimRight(client.HomeserverURL.String(), "/")
- downloadURL := base + "/_matrix/media/v3/download/" +
- url.PathEscape(server) + "/" + url.PathEscape(mediaID)
-
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
- if err != nil {
- return err
- }
- req.Header.Set("Authorization", "Bearer "+client.AccessToken)
- resp, err := http.DefaultClient.Do(req)
- if err != nil {
- return err
- }
- defer resp.Body.Close()
-
- if resp.StatusCode != http.StatusOK {
- return fmt.Errorf("media download HTTP %d for %s", resp.StatusCode, mxcURL)
- }
-
- data, err := io.ReadAll(resp.Body)
- if err != nil {
- return err
- }
-
- ct := resp.Header.Get("Content-Type")
- if ct == "" {
- ct = "application/octet-stream"
- }
-
- finalKey := s3Key
- if ext := extFromContentType(ct); ext != "" {
- finalKey = s3Key + ext
- }
- if err := s3Put(ctx, finalKey, data, ct); err != nil {
- return err
- }
- slog.Info("Stored media", "key", finalKey, "bytes", len(data))
- return nil
-}
-
-func extFromContentType(ct string) string {
- if i := strings.Index(ct, ";"); i >= 0 {
- ct = ct[:i]
- }
- switch strings.TrimSpace(ct) {
- case "image/jpeg":
- return ".jpg"
- case "image/png":
- return ".png"
- case "image/gif":
- return ".gif"
- case "image/webp":
- return ".webp"
- case "video/mp4":
- return ".mp4"
- case "video/webm":
- return ".webm"
- case "audio/mpeg":
- return ".mp3"
- case "audio/ogg":
- return ".ogg"
- case "audio/opus":
- return ".opus"
- case "application/pdf":
- return ".pdf"
- }
- return ""
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Event processing
-// ─────────────────────────────────────────────────────────────────────────────
-
-var mediaMsgTypes = map[string]bool{
- "m.image": true,
- "m.file": true,
- "m.video": true,
- "m.audio": true,
-}
-
-// processEvent downloads media and records profile changes for a single event.
-// The event should already be decrypted before this is called.
-func processEvent(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string, isDM bool) {
- if ev.Type == event.EventEncrypted {
- return
- }
-
- if ev.Type == event.EventMessage {
- var content struct {
- MsgType string `json:"msgtype"`
- URL string `json:"url"`
- File json.RawMessage `json:"file"`
- }
- if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil {
- mediaURL := content.URL
- if mediaURL == "" && content.File != nil {
- var f struct{ URL string `json:"url"` }
- if json.Unmarshal(content.File, &f) == nil {
- mediaURL = f.URL
- }
- }
- if mediaMsgTypes[content.MsgType] && mediaURL != "" {
- if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) {
- if err := downloadAndStoreMedia(ctx, client, mediaURL, prefix, "media"); err != nil {
- slog.Warn("Media download failed", "url", mediaURL, "error", err)
- }
- }
- }
- }
- }
-
- if ev.Type == event.EventSticker {
- var content struct {
- URL string `json:"url"`
- File json.RawMessage `json:"file"`
- }
- if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil {
- mediaURL := content.URL
- if mediaURL == "" && content.File != nil {
- var f struct{ URL string `json:"url"` }
- if json.Unmarshal(content.File, &f) == nil {
- mediaURL = f.URL
- }
- }
- if mediaURL != "" {
- if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) {
- if err := downloadAndStoreMedia(ctx, client, mediaURL, prefix, "media"); err != nil {
- slog.Warn("Sticker download failed", "url", mediaURL, "error", err)
- }
- }
- }
- }
- }
-
- if ev.Type == event.StateMember {
- recordProfileUpdate(ctx, client, ev, roomID, prefix)
- }
-}
-
-type profileUpdateRecord struct {
- TS int64 `json:"ts"`
- EventID string `json:"event_id"`
- UserID string `json:"user_id"`
- RoomID string `json:"room_id"`
- DisplayNameOld string `json:"displayname_old,omitempty"`
- DisplayNameNew string `json:"displayname_new,omitempty"`
- AvatarOld string `json:"avatar_old,omitempty"`
- AvatarNew string `json:"avatar_new,omitempty"`
-}
-
-// recordProfileUpdate appends a JSONL entry to S3 when a room member event shows
-// a display name or avatar change for a local user. Also downloads the new avatar.
-func recordProfileUpdate(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string) {
- if ev.StateKey == nil {
- return
- }
- stateKey := *ev.StateKey
- if !strings.HasSuffix(stateKey, ":"+ourServerName) {
- return
- }
- var content, prevContent struct {
- Displayname string `json:"displayname"`
- AvatarURL string `json:"avatar_url"`
- }
- if err := json.Unmarshal(ev.Content.VeryRaw, &content); err != nil {
- return
- }
- if ev.Unsigned.PrevContent != nil {
- _ = json.Unmarshal(ev.Unsigned.PrevContent.VeryRaw, &prevContent)
- }
- if content.Displayname == prevContent.Displayname && content.AvatarURL == prevContent.AvatarURL {
- return
- }
- rec := profileUpdateRecord{
- TS: ev.Timestamp,
- EventID: string(ev.ID),
- UserID: stateKey,
- RoomID: string(roomID),
- }
- if content.Displayname != prevContent.Displayname {
- rec.DisplayNameOld = prevContent.Displayname
- rec.DisplayNameNew = content.Displayname
- }
- if content.AvatarURL != prevContent.AvatarURL {
- rec.AvatarOld = prevContent.AvatarURL
- rec.AvatarNew = content.AvatarURL
- if content.AvatarURL != "" {
- if err := downloadAndStoreMedia(ctx, client, content.AvatarURL, prefix, "avatars"); err != nil {
- slog.Warn("Avatar download failed", "url", content.AvatarURL, "error", err)
- }
- }
- }
- line, err := json.Marshal(rec)
- if err != nil {
- return
- }
- s3Key := prefix + "/profile-updates.jsonl"
- existing, _ := s3Get(ctx, s3Key)
- _ = s3Put(ctx, s3Key, append(existing, append(line, '\n')...), "application/x-ndjson")
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// History pagination
-// ─────────────────────────────────────────────────────────────────────────────
-
-type historyEvent struct {
- EventID string `json:"event_id"`
- Sender string `json:"sender"`
- Type string `json:"type"`
- Timestamp int64 `json:"origin_server_ts"`
- Content json.RawMessage `json:"content"`
- Encrypted bool `json:"encrypted,omitempty"`
-}
-
-func eventToRecord(ev *event.Event, wasEncrypted bool) historyEvent {
- return historyEvent{
- EventID: string(ev.ID),
- Sender: string(ev.Sender),
- Type: ev.Type.Type,
- Timestamp: ev.Timestamp,
- Content: ev.Content.VeryRaw,
- Encrypted: wasEncrypted,
- }
-}
-
-// paginateRoom fetches room history and writes new events to a per-run
-// age-encrypted JSONL file in S3. For DMs on first run, the full history is
-// fetched backwards from the current position before normal forward pagination.
-//
-// syncNextBatch is the nextBatch token from the current sync response. On first
-// run we save it as the forward cursor so the next run only sees events that
-// arrive after this sync — avoiding re-fetching the events the sync already
-// returned.
-func paginateRoom(ctx context.Context, client *mautrix.Client, roomID id.RoomID, prefix string, isDM bool, prevBatch string, syncNextBatch string, sessions megolmSessions) error {
- safeKey := strings.NewReplacer("/", "_", ":", "_").Replace(string(roomID))
- cursorKey := prefix + "/history-cursor/" + safeKey + ".json"
- dmHistoryDoneKey := prefix + "/history-cursor/" + safeKey + ".dm-done"
-
- var cursor struct {
- Token string `json:"token"`
- }
- if err := s3GetJSON(ctx, cursorKey, &cursor); err != nil {
- return err
- }
-
- // fetchDMHistory does a full backward walk from prevBatch and stores results.
- // Used on first-run for DMs, and also as a catch-up when a room was initially
- // missed (e.g. getDMRooms bug meant isDM=false on first run).
- fetchDMHistory := func(fromToken string) {
- slog.Info("DM: fetching full history", "room_id", roomID, "from_token", fromToken)
- var messages []historyEvent
- token := fromToken
- for {
- resp, err := client.Messages(ctx, roomID, token, "", mautrix.DirectionBackward, nil, 100)
- if err != nil {
- slog.Warn("room_messages error (backward)", "room_id", roomID, "token", token, "error", err)
- break
- }
- slog.Info("DM history page", "room_id", roomID, "chunk_size", len(resp.Chunk), "end", resp.End)
- if len(resp.Chunk) == 0 {
- break
- }
- for _, ev := range resp.Chunk {
- wasEncrypted := ev.Type == event.EventEncrypted
- ev = tryDecryptEvent(ev, sessions)
- messages = append(messages, eventToRecord(ev, wasEncrypted))
- processEvent(ctx, client, ev, roomID, prefix, true)
- }
- if resp.End == "" || resp.End == token {
- break
- }
- token = resp.End
- }
- if len(messages) > 0 {
- for i, j := 0, len(messages)-1; i < j; i, j = i+1, j-1 {
- messages[i], messages[j] = messages[j], messages[i]
- }
- histKey := prefix + "/history/" + safeKey + "/" + runStr + "-backfill.jsonl"
- var buf bytes.Buffer
- for _, m := range messages {
- line, _ := json.Marshal(m)
- buf.Write(line)
- buf.WriteByte('\n')
- }
- if err := s3PutAge(ctx, histKey, buf.Bytes()); err != nil {
- slog.Warn("Failed to upload DM history", "room_id", roomID, "error", err)
- } else {
- slog.Info("DM history stored", "room_id", roomID, "events", len(messages))
- _ = s3Put(ctx, dmHistoryDoneKey, []byte("done"), "text/plain")
- }
- } else {
- // Nothing to backfill — mark as done so we don't retry every run.
- slog.Info("DM history: no messages found, marking done", "room_id", roomID)
- if err := s3Put(ctx, dmHistoryDoneKey, []byte("done"), "text/plain"); err != nil {
- slog.Warn("Failed to write dm-done marker", "room_id", roomID, "error", err)
- }
- }
- }
-
- if cursor.Token == "" {
- // First time we've seen this room. For DMs, walk all the way back through
- // history so we have a complete record from day one.
- if isDM && prevBatch != "" {
- fetchDMHistory(prevBatch)
- }
- // Save the sync's nextBatch as the forward cursor. This means the next
- // run will forward-paginate from here and only pick up genuinely new
- // events, not the batch we already received in this sync response.
- if syncNextBatch == "" {
- syncNextBatch = prevBatch
- }
- if syncNextBatch != "" {
- cursorJSON, _ := json.Marshal(map[string]string{"token": syncNextBatch})
- _ = s3Put(ctx, cursorKey, cursorJSON, "application/json")
- }
- return nil
- }
-
- // Incremental run: if this is a DM but we never completed a backward history
- // fetch (e.g. room was missed because it wasn't in m.direct on first run),
- // do it now using the saved cursor as the start for backward pagination.
- // Note: prevBatch may be empty for rooms with no new events in a full_state
- // sync, so we use cursor.Token directly rather than gating on prevBatch.
- if isDM {
- dmDoneData, _ := s3Get(ctx, dmHistoryDoneKey)
- if dmDoneData == nil {
- slog.Info("DM catch-up: no history marker found, backfilling", "room_id", roomID)
- fetchDMHistory(cursor.Token)
- }
- }
-
- // Incremental run: fetch every page forward from the saved cursor until we
- // reach the live end of the timeline.
- var messages []historyEvent
- nextToken := cursor.Token
- lastGoodToken := cursor.Token
- histKey := prefix + "/history/" + safeKey + "/" + runStr + ".jsonl"
-
- for range 100 {
- resp, err := client.Messages(ctx, roomID, nextToken, "", mautrix.DirectionForward, nil, 100)
- if err != nil {
- slog.Warn("room_messages error (forward)", "room_id", roomID, "error", err)
- break
- }
- if len(resp.Chunk) == 0 {
- // No events — we are already at the live end.
- break
- }
- for _, ev := range resp.Chunk {
- wasEncrypted := ev.Type == event.EventEncrypted
- ev = tryDecryptEvent(ev, sessions)
- messages = append(messages, eventToRecord(ev, wasEncrypted))
- processEvent(ctx, client, ev, roomID, prefix, isDM)
- }
- if resp.End == "" || resp.End == nextToken {
- // Reached the live end of the timeline; nextToken is the furthest
- // position we can name.
- lastGoodToken = nextToken
- nextToken = ""
- break
- }
- lastGoodToken = resp.End
- nextToken = resp.End
- }
- // If the loop hit the 100-page cap before reaching live end, nextToken still
- // holds the resume position for the next run.
- if nextToken != "" {
- lastGoodToken = nextToken
- }
-
- if len(messages) > 0 {
- var buf bytes.Buffer
- for _, m := range messages {
- line, _ := json.Marshal(m)
- buf.Write(line)
- buf.WriteByte('\n')
- }
- if err := s3PutAge(ctx, histKey, buf.Bytes()); err != nil {
- slog.Warn("Failed to upload history", "room_id", roomID, "error", err)
- }
- slog.Info("History updated", "room_id", roomID, "new_events", len(messages))
- }
-
- // Always advance the cursor so the next run never re-processes events we've
- // already stored.
- if lastGoodToken != cursor.Token {
- cursorJSON, _ := json.Marshal(map[string]string{"token": lastGoodToken})
- _ = s3Put(ctx, cursorKey, cursorJSON, "application/json")
- }
- return nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// Per-account backup
-// ─────────────────────────────────────────────────────────────────────────────
-
-// backupAccount runs the full backup pipeline for a single Matrix account:
-// restore state, sync, export E2EE keys, save room list, paginate history,
-// then persist state back to S3.
-func backupAccount(ctx context.Context, acc accountCfg) error {
- slog.Info("=== Backing up account ===", "user_id", acc.UserID)
-
- storeS3Key := acc.Prefix + "/crypto-store.tar.gz"
- sessionS3Key := acc.Prefix + "/session.json"
- syncTokenS3Key := acc.Prefix + "/sync-token.json"
-
- if err := downloadStore(ctx, acc.StoreDir, storeS3Key); err != nil {
- slog.Warn("Could not restore store", "error", err)
- }
-
- client, err := mautrix.NewClient(homeserver, "", "")
- if err != nil {
- return fmt.Errorf("create client: %w", err)
- }
-
- if _, err := ensureSession(ctx, client, acc, sessionS3Key); err != nil {
- return fmt.Errorf("session: %w", err)
- }
-
- var syncToken struct {
- NextBatch string `json:"next_batch"`
- }
- _ = s3GetJSON(ctx, syncTokenS3Key, &syncToken)
-
- slog.Info("Syncing", "since", syncToken.NextBatch)
- syncResp, err := client.SyncRequest(ctx, 60000, syncToken.NextBatch, "", true, event.PresenceUnavailable)
- if err != nil {
- return fmt.Errorf("sync: %w", err)
- }
- if data, err := json.Marshal(map[string]string{"next_batch": syncResp.NextBatch}); err == nil {
- _ = s3Put(ctx, syncTokenS3Key, data, "application/json")
- }
- slog.Info("Sync done", "rooms", len(syncResp.Rooms.Join))
-
- var sessions megolmSessions
- if acc.SSSSKey != "" {
- var err error
- sessions, err = fetchAndExportKeyBackup(ctx, client, acc.SSSSKey, acc.Prefix)
- if err != nil {
- slog.Warn("Key backup export failed", "error", err)
- }
- }
-
- if err := saveRoomList(ctx, client, syncResp, acc.Prefix); err != nil {
- slog.Warn("Room list failed", "error", err)
- }
-
- dmRooms := getDMRooms(ctx, client, syncResp)
- // Augment dmRooms with rooms marked as "dm" in the stored rooms list.
- // This catches rooms classified as DMs by the 2-member heuristic in
- // saveRoomList that aren't in m.direct (e.g. created by the other party).
- if raw, err := getDecryptedAgeFromS3(ctx, acc.Prefix+"/rooms-latest.json.age"); err == nil && raw != nil {
- var storedRooms []roomEntry
- if json.Unmarshal(raw, &storedRooms) == nil {
- for _, r := range storedRooms {
- if r.Type == "dm" {
- dmRooms[id.RoomID(r.RoomID)] = true
- }
- }
- }
- }
- for roomID, joinedRoom := range syncResp.Rooms.Join {
- isDM := dmRooms[roomID]
- if err := paginateRoom(ctx, client, roomID, acc.Prefix, isDM, joinedRoom.Timeline.PrevBatch, syncResp.NextBatch, sessions); err != nil {
- slog.Warn("History error", "room_id", roomID, "error", err)
- }
- }
-
- // Catch-up: DM rooms absent from this sync delta (no new events) may still
- // need their history backfilled. Use the rooms list just written by saveRoomList
- // (which also detects 2-member rooms as DMs) as the authoritative DM source,
- // falling back to the API-based dmRooms map.
- catchupDMs := make(map[id.RoomID]bool)
- for roomID := range dmRooms {
- catchupDMs[roomID] = true
- }
- if raw, err := getDecryptedAgeFromS3(ctx, acc.Prefix+"/rooms-latest.json.age"); err != nil {
- slog.Warn("DM catch-up: failed to read rooms list", "error", err)
- } else if raw == nil {
- slog.Warn("DM catch-up: rooms list not found or ageIdentity not set")
- } else {
- var storedRooms []roomEntry
- if err := json.Unmarshal(raw, &storedRooms); err != nil {
- slog.Warn("DM catch-up: failed to parse rooms list", "error", err)
- } else {
- slog.Info("DM catch-up: loaded rooms list", "total_rooms", len(storedRooms))
- for _, r := range storedRooms {
- if r.Type == "dm" {
- catchupDMs[id.RoomID(r.RoomID)] = true
- }
- }
- }
- }
- slog.Info("DM catch-up: checking rooms", "total_dm_rooms", len(catchupDMs), "in_sync", len(syncResp.Rooms.Join))
- for roomID := range catchupDMs {
- if _, inSync := syncResp.Rooms.Join[roomID]; inSync {
- continue // already handled above
- }
- safeKey := strings.NewReplacer("/", "_", ":", "_").Replace(string(roomID))
- dmHistoryDoneKey := acc.Prefix + "/history-cursor/" + safeKey + ".dm-done"
- dmDone, dmDoneErr := s3Get(ctx, dmHistoryDoneKey)
- if dmDoneErr != nil {
- slog.Warn("DM catch-up: error checking dm-done marker", "room_id", roomID, "error", dmDoneErr)
- }
- if dmDone != nil {
- continue // already backfilled
- }
- cursorKey := acc.Prefix + "/history-cursor/" + safeKey + ".json"
- var cursor struct {
- Token string `json:"token"`
- }
- if err := s3GetJSON(ctx, cursorKey, &cursor); err != nil {
- slog.Warn("DM catch-up: error reading cursor", "room_id", roomID, "error", err)
- continue
- }
- if cursor.Token == "" {
- slog.Info("DM catch-up: no cursor yet, skipping", "room_id", roomID)
- continue
- }
- slog.Info("DM catch-up: backfilling room", "room_id", roomID, "cursor", cursor.Token)
- if err := paginateRoom(ctx, client, roomID, acc.Prefix, true, cursor.Token, syncResp.NextBatch, sessions); err != nil {
- slog.Warn("DM catch-up error", "room_id", roomID, "error", err)
- }
- }
-
- if err := uploadStore(ctx, acc.StoreDir, storeS3Key); err != nil {
- slog.Warn("Could not upload store", "error", err)
- }
-
- slog.Info("Done", "user_id", acc.UserID)
- return nil
-}
-
-// ─────────────────────────────────────────────────────────────────────────────
-// main
-// ─────────────────────────────────────────────────────────────────────────────
-
func main() {
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
Level: slog.LevelInfo,
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/matrix.go b/apps/talos_cluster/matrix-backup/backup-tool/matrix.go
@@ -0,0 +1,47 @@
+// matrix.go — low-level Matrix client helpers not wrapped by mautrix.
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "strings"
+
+ "maunium.net/go/mautrix"
+)
+
+// matrixGetJSON makes an authenticated GET to the homeserver and JSON-decodes
+// the response. Used for endpoints not wrapped by the mautrix client.
+func matrixGetJSON(ctx context.Context, client *mautrix.Client, path string, out interface{}) error {
+ base := strings.TrimRight(client.HomeserverURL.String(), "/")
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+path, nil)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Authorization", "Bearer "+client.AccessToken)
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return err
+ }
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("HTTP %d: %s", resp.StatusCode, body)
+ }
+ return json.Unmarshal(body, out)
+}
+
+// getAccountData fetches a single account-data event for the authenticated user.
+func getAccountData(ctx context.Context, client *mautrix.Client, eventType string, out interface{}) error {
+ path := "/_matrix/client/v3/user/" +
+ url.PathEscape(client.UserID.String()) +
+ "/account_data/" +
+ url.PathEscape(eventType)
+ return matrixGetJSON(ctx, client, path, out)
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/media.go b/apps/talos_cluster/matrix-backup/backup-tool/media.go
@@ -0,0 +1,117 @@
+// media.go — Matrix media download and S3 storage.
+package main
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/url"
+ "strings"
+ "time"
+
+ "maunium.net/go/mautrix"
+)
+
+// mediaHTTPClient is used for all media downloads; the 5-minute timeout
+// prevents a single large file from stalling the entire backup run.
+var mediaHTTPClient = &http.Client{Timeout: 5 * time.Minute}
+
+var mediaMsgTypes = map[string]bool{
+ "m.image": true,
+ "m.file": true,
+ "m.video": true,
+ "m.audio": true,
+}
+
+// downloadAndStoreMedia downloads a single mxc:// URL and stores it under
+// prefix/label/server/mediaID in S3, skipping if already present.
+func downloadAndStoreMedia(ctx context.Context, client *mautrix.Client, mxcURL, prefix, label string) error {
+ if !strings.HasPrefix(mxcURL, "mxc://") {
+ return nil
+ }
+ rest := mxcURL[len("mxc://"):]
+ slashIdx := strings.Index(rest, "/")
+ if slashIdx < 0 {
+ return nil
+ }
+ server := rest[:slashIdx]
+ mediaID := rest[slashIdx+1:]
+ if server == "" || mediaID == "" {
+ return nil
+ }
+
+ s3Key := prefix + "/" + label + "/" + server + "/" + mediaID
+ if s3Exists(ctx, s3Key) {
+ return nil
+ }
+
+ base := strings.TrimRight(client.HomeserverURL.String(), "/")
+ downloadURL := base + "/_matrix/media/v3/download/" +
+ url.PathEscape(server) + "/" + url.PathEscape(mediaID)
+
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Authorization", "Bearer "+client.AccessToken)
+ resp, err := mediaHTTPClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("media download HTTP %d for %s", resp.StatusCode, mxcURL)
+ }
+
+ data, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return err
+ }
+
+ ct := resp.Header.Get("Content-Type")
+ if ct == "" {
+ ct = "application/octet-stream"
+ }
+
+ finalKey := s3Key
+ if ext := extFromContentType(ct); ext != "" {
+ finalKey = s3Key + ext
+ }
+ if err := s3Put(ctx, finalKey, data, ct); err != nil {
+ return err
+ }
+ slog.Info("Stored media", "key", finalKey, "bytes", len(data))
+ return nil
+}
+
+func extFromContentType(ct string) string {
+ if i := strings.Index(ct, ";"); i >= 0 {
+ ct = ct[:i]
+ }
+ switch strings.TrimSpace(ct) {
+ case "image/jpeg":
+ return ".jpg"
+ case "image/png":
+ return ".png"
+ case "image/gif":
+ return ".gif"
+ case "image/webp":
+ return ".webp"
+ case "video/mp4":
+ return ".mp4"
+ case "video/webm":
+ return ".webm"
+ case "audio/mpeg":
+ return ".mp3"
+ case "audio/ogg":
+ return ".ogg"
+ case "audio/opus":
+ return ".opus"
+ case "application/pdf":
+ return ".pdf"
+ }
+ return ""
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/rooms.go b/apps/talos_cluster/matrix-backup/backup-tool/rooms.go
@@ -0,0 +1,356 @@
+// rooms.go — room list management: name calculation, DM detection, S3 persistence.
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "log/slog"
+ "strings"
+
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/event"
+ "maunium.net/go/mautrix/id"
+ "net/url"
+)
+
+type roomEntry struct {
+ RoomID string `json:"room_id"`
+ Name string `json:"name"`
+ Type string `json:"type"`
+ Aliases []string `json:"aliases"`
+ CanonicalAlias string `json:"canonical_alias,omitempty"`
+ MemberCount int `json:"member_count"`
+ Encrypted bool `json:"encrypted"`
+ ViaServers []string `json:"via_servers,omitempty"`
+}
+
+// calcRoomName implements the Matrix room display-name algorithm:
+// explicit name → canonical alias → heroes → "Empty Room".
+func calcRoomName(
+ stateEvents []*event.Event,
+ heroes []id.UserID,
+ joinedCount, invitedCount int,
+ selfUserID id.UserID,
+) string {
+ for _, ev := range stateEvents {
+ if ev.Type == event.StateRoomName {
+ if c := ev.Content.AsRoomName(); c != nil && c.Name != "" {
+ return c.Name
+ }
+ }
+ }
+ for _, ev := range stateEvents {
+ if ev.Type == event.StateCanonicalAlias {
+ if c := ev.Content.AsCanonicalAlias(); c != nil && c.Alias != "" {
+ return string(c.Alias)
+ }
+ }
+ }
+
+ memberNames := make(map[id.UserID]string)
+ for _, ev := range stateEvents {
+ if ev.Type == event.StateMember && ev.StateKey != nil {
+ uid := id.UserID(*ev.StateKey)
+ if c := ev.Content.AsMember(); c != nil && c.Displayname != "" {
+ memberNames[uid] = c.Displayname
+ } else {
+ local, _, _ := uid.Parse()
+ if local != "" {
+ memberNames[uid] = local
+ } else {
+ memberNames[uid] = string(uid)
+ }
+ }
+ }
+ }
+
+ var filtered []id.UserID
+ for _, h := range heroes {
+ if h != selfUserID {
+ filtered = append(filtered, h)
+ }
+ }
+
+ tentativeNames := make(map[id.UserID]string, len(filtered))
+ for _, h := range filtered {
+ if n, ok := memberNames[h]; ok {
+ tentativeNames[h] = n
+ } else {
+ local, _, _ := h.Parse()
+ if local != "" {
+ tentativeNames[h] = local
+ } else {
+ tentativeNames[h] = string(h)
+ }
+ }
+ }
+ nameCounts := make(map[string]int, len(filtered))
+ for _, n := range tentativeNames {
+ nameCounts[n]++
+ }
+ heroName := func(uid id.UserID) string {
+ n := tentativeNames[uid]
+ if nameCounts[n] > 1 {
+ _, server, err := uid.Parse()
+ if err == nil && server != "" {
+ return n + " (" + server + ")"
+ }
+ return string(uid)
+ }
+ return n
+ }
+
+ others := joinedCount + invitedCount - 1
+ if others < 0 {
+ others = 0
+ }
+ if len(filtered) == 0 {
+ if others == 0 {
+ return "Empty Room"
+ }
+ return ""
+ }
+
+ names := make([]string, len(filtered))
+ for i, h := range filtered {
+ names[i] = heroName(h)
+ }
+ unnamed := others - len(names)
+ if unnamed <= 0 {
+ switch len(names) {
+ case 1:
+ return names[0]
+ case 2:
+ return names[0] + " and " + names[1]
+ default:
+ return strings.Join(names[:len(names)-1], ", ") + ", and " + names[len(names)-1]
+ }
+ }
+ switch len(names) {
+ case 1:
+ return fmt.Sprintf("%s and %d others", names[0], unnamed)
+ case 2:
+ return fmt.Sprintf("%s, %s, and %d others", names[0], names[1], unnamed)
+ default:
+ return fmt.Sprintf("%s, and %d others", strings.Join(names, ", "), unnamed)
+ }
+}
+
+// collectViaServers returns up to 3 server names for matrix.to ?via= links.
+func collectViaServers(roomID id.RoomID, heroes []id.UserID, stateEvents []*event.Event) []string {
+ seen := make(map[string]bool)
+ var servers []string
+ add := func(s string) bool {
+ if s != "" && !seen[s] {
+ seen[s] = true
+ servers = append(servers, s)
+ }
+ return len(servers) >= 3
+ }
+ if idx := strings.LastIndex(string(roomID), ":"); idx >= 0 {
+ add(string(roomID)[idx+1:])
+ }
+ for _, h := range heroes {
+ _, server, err := h.Parse()
+ if err == nil && add(server) {
+ return servers
+ }
+ }
+ for _, ev := range stateEvents {
+ if len(servers) >= 3 {
+ break
+ }
+ if ev.Type != event.StateMember || ev.StateKey == nil {
+ continue
+ }
+ var mc struct {
+ Membership string `json:"membership"`
+ }
+ if ev.Content.VeryRaw != nil {
+ _ = json.Unmarshal(ev.Content.VeryRaw, &mc)
+ }
+ if mc.Membership != "join" {
+ continue
+ }
+ _, server, err := id.UserID(*ev.StateKey).Parse()
+ if err == nil {
+ add(server)
+ }
+ }
+ return servers
+}
+
+// fetchRoomNameFromState fetches the room name directly from the state API.
+// Used when the sync delta doesn't include the name event.
+func fetchRoomNameFromState(ctx context.Context, client *mautrix.Client, roomID id.RoomID) string {
+ base := url.PathEscape(string(roomID))
+ var nameContent struct {
+ Name string `json:"name"`
+ }
+ if err := matrixGetJSON(ctx, client,
+ "/_matrix/client/v3/rooms/"+base+"/state/m.room.name",
+ &nameContent,
+ ); err == nil && nameContent.Name != "" {
+ return nameContent.Name
+ }
+ var aliasContent struct {
+ Alias string `json:"alias"`
+ }
+ if err := matrixGetJSON(ctx, client,
+ "/_matrix/client/v3/rooms/"+base+"/state/m.room.canonical_alias",
+ &aliasContent,
+ ); err == nil && aliasContent.Alias != "" {
+ return aliasContent.Alias
+ }
+ return ""
+}
+
+// saveRoomList builds a JSON snapshot of all joined rooms and uploads it to S3.
+// Rooms absent from this sync delta are preserved from the prior list.
+func saveRoomList(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync, prefix string) error {
+ existing := make(map[string]roomEntry)
+ if raw, err := getDecryptedAgeFromS3(ctx, prefix+"/rooms-latest.json.age"); err != nil {
+ slog.Warn("Could not load previous room list", "error", err)
+ } else if raw != nil {
+ var prev []roomEntry
+ if err := json.Unmarshal(raw, &prev); err != nil {
+ slog.Warn("Previous room list is corrupt, starting fresh", "error", err)
+ } else {
+ for _, r := range prev {
+ // Retroactively reclassify 2-member non-space rooms stored as
+ // "normal" before the member-count heuristic was introduced.
+ if r.Type == "normal" && r.MemberCount > 0 && r.MemberCount <= 2 {
+ r.Type = "dm"
+ }
+ existing[r.RoomID] = r
+ }
+ }
+ }
+
+ dmRooms := getDMRooms(ctx, client, syncResp)
+
+ for roomID, joinedRoom := range syncResp.Rooms.Join {
+ var name, canonicalAlias string
+ var aliases []string
+ encrypted := false
+ rtype := "normal"
+ if dmRooms[roomID] {
+ rtype = "dm"
+ }
+
+ for _, ev := range joinedRoom.State.Events {
+ _ = ev.Content.ParseRaw(ev.Type)
+ }
+ for _, ev := range joinedRoom.State.Events {
+ switch ev.Type {
+ case event.StateRoomName:
+ if c := ev.Content.AsRoomName(); c.Name != "" {
+ name = c.Name
+ }
+ case event.StateCanonicalAlias:
+ if c := ev.Content.AsCanonicalAlias(); c.Alias != "" {
+ canonicalAlias = string(c.Alias)
+ for _, a := range c.AltAliases {
+ aliases = append(aliases, string(a))
+ }
+ }
+ case event.StateEncryption:
+ encrypted = true
+ case event.StateCreate:
+ if ev.Content.VeryRaw != nil {
+ var createContent struct {
+ Type string `json:"type"`
+ }
+ if json.Unmarshal(ev.Content.VeryRaw, &createContent) == nil && createContent.Type == "m.space" {
+ rtype = "space"
+ }
+ }
+ }
+ }
+
+ joinedCount, invitedCount := 0, 0
+ if joinedRoom.Summary.JoinedMemberCount != nil {
+ joinedCount = *joinedRoom.Summary.JoinedMemberCount
+ }
+ if joinedRoom.Summary.InvitedMemberCount != nil {
+ invitedCount = *joinedRoom.Summary.InvitedMemberCount
+ }
+ if rtype == "normal" && joinedCount > 0 && joinedCount <= 2 {
+ rtype = "dm"
+ }
+
+ if name == "" {
+ name = calcRoomName(joinedRoom.State.Events, joinedRoom.Summary.Heroes, joinedCount, invitedCount, client.UserID)
+ }
+ if name == "" {
+ name = fetchRoomNameFromState(ctx, client, roomID)
+ }
+ if name == "" {
+ if prev, ok := existing[string(roomID)]; ok {
+ name = prev.Name
+ }
+ }
+ if name == "" {
+ name = string(roomID)
+ }
+
+ existing[string(roomID)] = roomEntry{
+ RoomID: string(roomID),
+ Name: name,
+ Type: rtype,
+ Aliases: aliases,
+ CanonicalAlias: canonicalAlias,
+ MemberCount: joinedCount,
+ Encrypted: encrypted,
+ ViaServers: collectViaServers(roomID, joinedRoom.Summary.Heroes, joinedRoom.State.Events),
+ }
+ slog.Info("Room", "type", rtype, "name", name)
+ }
+
+ rooms := make([]roomEntry, 0, len(existing))
+ for _, r := range existing {
+ rooms = append(rooms, r)
+ }
+ data, err := json.MarshalIndent(rooms, "", " ")
+ if err != nil {
+ return err
+ }
+ if err := s3PutAge(ctx, prefix+"/rooms-"+dateStr+".json", data); err != nil {
+ return err
+ }
+ if err := s3PutAge(ctx, prefix+"/rooms-latest.json", data); err != nil {
+ return err
+ }
+ slog.Info("Uploaded room list", "rooms", len(rooms))
+ return nil
+}
+
+// getDMRooms returns the full set of room IDs marked as direct chats.
+// Always fetches from the account-data API so incremental syncs (which only
+// carry changed account-data events) don't see a stale map.
+func getDMRooms(ctx context.Context, client *mautrix.Client, syncResp *mautrix.RespSync) map[id.RoomID]bool {
+ dmRooms := make(map[id.RoomID]bool)
+ populate := func(direct event.DirectChatsEventContent) {
+ for _, roomIDs := range direct {
+ for _, rid := range roomIDs {
+ dmRooms[rid] = true
+ }
+ }
+ }
+ var direct event.DirectChatsEventContent
+ if err := getAccountData(ctx, client, "m.direct", &direct); err == nil {
+ populate(direct)
+ return dmRooms
+ }
+ // Fallback: use whatever the sync included.
+ for _, ev := range syncResp.AccountData.Events {
+ if ev.Type == event.AccountDataDirectChats {
+ if err := json.Unmarshal(ev.Content.VeryRaw, &direct); err == nil {
+ populate(direct)
+ }
+ break
+ }
+ }
+ return dmRooms
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/s3.go b/apps/talos_cluster/matrix-backup/backup-tool/s3.go
@@ -0,0 +1,154 @@
+// s3.go — S3 client initialisation, helpers, and age encryption/decryption.
+package main
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+
+ "filippo.io/age"
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ "github.com/aws/aws-sdk-go-v2/service/s3"
+ s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+)
+
+var s3c *s3.Client
+
+func initS3(ctx context.Context) error {
+ cfg, err := awsconfig.LoadDefaultConfig(ctx,
+ awsconfig.WithRegion(s3Region),
+ awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
+ s3AccessKey, s3SecretKey, "",
+ )),
+ )
+ if err != nil {
+ return err
+ }
+ endpoint := "https://" + s3Endpoint
+ s3c = s3.NewFromConfig(cfg, func(o *s3.Options) {
+ o.UsePathStyle = true
+ o.BaseEndpoint = &endpoint
+ })
+ return nil
+}
+
+func s3Get(ctx context.Context, key string) ([]byte, error) {
+ out, err := s3c.GetObject(ctx, &s3.GetObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: aws.String(key),
+ })
+ if err != nil {
+ var nsk *s3types.NoSuchKey
+ if errors.As(err, &nsk) {
+ return nil, nil
+ }
+ return nil, err
+ }
+ defer out.Body.Close()
+ return io.ReadAll(out.Body)
+}
+
+func s3GetJSON(ctx context.Context, key string, out interface{}) error {
+ data, err := s3Get(ctx, key)
+ if err != nil || data == nil {
+ return err
+ }
+ return json.Unmarshal(data, out)
+}
+
+func s3Put(ctx context.Context, key string, data []byte, contentType string) error {
+ _, err := s3c.PutObject(ctx, &s3.PutObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: aws.String(key),
+ Body: bytes.NewReader(data),
+ ContentType: aws.String(contentType),
+ })
+ return err
+}
+
+func s3Exists(ctx context.Context, key string) bool {
+ _, err := s3c.HeadObject(ctx, &s3.HeadObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: aws.String(key),
+ })
+ return err == nil
+}
+
+// s3DeletePrefix deletes all objects whose key starts with prefix.
+func s3DeletePrefix(ctx context.Context, prefix string) error {
+ paginator := s3.NewListObjectsV2Paginator(s3c, &s3.ListObjectsV2Input{
+ Bucket: aws.String(s3BucketName),
+ Prefix: aws.String(prefix),
+ })
+ deleted := 0
+ for paginator.HasMorePages() {
+ page, err := paginator.NextPage(ctx)
+ if err != nil {
+ return err
+ }
+ for _, obj := range page.Contents {
+ if _, err := s3c.DeleteObject(ctx, &s3.DeleteObjectInput{
+ Bucket: aws.String(s3BucketName),
+ Key: obj.Key,
+ }); err != nil {
+ slog.Warn("Failed to delete S3 object", "key", *obj.Key, "error", err)
+ } else {
+ deleted++
+ }
+ }
+ }
+ slog.Info("Deleted S3 prefix", "prefix", prefix, "count", deleted)
+ return nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Age encryption / decryption
+// ─────────────────────────────────────────────────────────────────────────────
+
+// ageEncrypt encrypts data for all ageRecipients and returns the ciphertext.
+func ageEncrypt(data []byte) ([]byte, error) {
+ var buf bytes.Buffer
+ w, err := age.Encrypt(&buf, ageRecipients...)
+ if err != nil {
+ return nil, err
+ }
+ if _, err := w.Write(data); err != nil {
+ return nil, err
+ }
+ if err := w.Close(); err != nil {
+ return nil, err
+ }
+ return buf.Bytes(), nil
+}
+
+// getDecryptedAgeFromS3 fetches an .age object from S3 and decrypts it.
+// Returns (nil, nil) when the object doesn't exist or no identity is available.
+func getDecryptedAgeFromS3(ctx context.Context, key string) ([]byte, error) {
+ if ageIdentity == nil {
+ return nil, nil
+ }
+ enc, err := s3Get(ctx, key)
+ if err != nil || enc == nil {
+ return nil, err
+ }
+ r, err := age.Decrypt(bytes.NewReader(enc), ageIdentity)
+ if err != nil {
+ return nil, fmt.Errorf("age decrypt %s: %w", key, err)
+ }
+ return io.ReadAll(r)
+}
+
+// s3PutAge age-encrypts data then uploads it under key+".age".
+func s3PutAge(ctx context.Context, key string, data []byte) error {
+ enc, err := ageEncrypt(data)
+ if err != nil {
+ return fmt.Errorf("age encrypt %s: %w", key, err)
+ }
+ return s3Put(ctx, key+".age", enc, "application/octet-stream")
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/session.go b/apps/talos_cluster/matrix-backup/backup-tool/session.go
@@ -0,0 +1,81 @@
+// session.go — Matrix session persistence (login, save, restore).
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "log/slog"
+
+ "maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/id"
+)
+
+type sessionData struct {
+ AccessToken string `json:"access_token"`
+ DeviceID id.DeviceID `json:"device_id"`
+}
+
+// ensureSession loads a saved session from S3 or logs in with the account
+// password and persists the new session for future runs.
+//
+// Sessions are stored age-encrypted under s3Key+".age". Plain-JSON sessions
+// written by older tool versions are transparently read as a fallback so
+// existing deployments migrate automatically on the next run.
+func ensureSession(ctx context.Context, client *mautrix.Client, acc accountCfg, s3Key string) (*sessionData, error) {
+ var sess sessionData
+
+ // Try age-encrypted first (current format).
+ if raw, err := getDecryptedAgeFromS3(ctx, s3Key+".age"); err != nil {
+ return nil, fmt.Errorf("load session: %w", err)
+ } else if raw != nil {
+ if err := json.Unmarshal(raw, &sess); err != nil {
+ return nil, fmt.Errorf("parse session: %w", err)
+ }
+ } else {
+ // Fallback: plain JSON written by older versions.
+ if err := s3GetJSON(ctx, s3Key, &sess); err != nil {
+ return nil, fmt.Errorf("load session (legacy): %w", err)
+ }
+ }
+
+ if sess.AccessToken != "" {
+ slog.Info("Restored existing session", "device_id", sess.DeviceID)
+ client.UserID = acc.UserID
+ client.DeviceID = sess.DeviceID
+ client.AccessToken = sess.AccessToken
+ return &sess, nil
+ }
+
+ slog.Info("First run — logging in with password", "user_id", acc.UserID)
+ resp, err := client.Login(ctx, &mautrix.ReqLogin{
+ Type: mautrix.AuthTypePassword,
+ Identifier: mautrix.UserIdentifier{
+ Type: mautrix.IdentifierTypeUser,
+ User: string(acc.UserID),
+ },
+ Password: acc.Password,
+ InitialDeviceDisplayName: "matrix-backup",
+ })
+ if err != nil {
+ return nil, fmt.Errorf("login: %w", err)
+ }
+ sess = sessionData{
+ AccessToken: resp.AccessToken,
+ DeviceID: resp.DeviceID,
+ }
+ client.UserID = resp.UserID
+ client.DeviceID = resp.DeviceID
+ client.AccessToken = resp.AccessToken
+ slog.Info("Logged in", "device_id", resp.DeviceID)
+
+ data, err := json.Marshal(sess)
+ if err != nil {
+ return nil, err
+ }
+ // Write age-encrypted; appends ".age" to s3Key.
+ if err := s3PutAge(ctx, s3Key, data); err != nil {
+ return nil, fmt.Errorf("save session: %w", err)
+ }
+ return &sess, nil
+}
diff --git a/apps/talos_cluster/matrix-backup/backup-tool/store.go b/apps/talos_cluster/matrix-backup/backup-tool/store.go
@@ -0,0 +1,134 @@
+// store.go — crypto-store tarball backup and restore (persists Megolm sessions
+// and sync tokens across CronJob runs via S3).
+package main
+
+import (
+ "archive/tar"
+ "bytes"
+ "compress/gzip"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "os"
+)
+
+// downloadStore restores a previously uploaded crypto store tarball from S3.
+// A missing store (first run) is not an error — the caller starts fresh.
+func downloadStore(ctx context.Context, storeDir, s3Key string) error {
+ data, err := s3Get(ctx, s3Key)
+ if err != nil {
+ return err
+ }
+ if data == nil {
+ slog.Info("No existing store in S3, starting fresh", "key", s3Key)
+ return nil
+ }
+ if err := os.MkdirAll(storeDir, 0700); err != nil {
+ return err
+ }
+ gr, err := gzip.NewReader(bytes.NewReader(data))
+ if err != nil {
+ return err
+ }
+ defer gr.Close()
+ tr := tar.NewReader(gr)
+ for {
+ hdr, err := tr.Next()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return err
+ }
+ target := storeDir + "/" + hdr.Name
+ switch hdr.Typeflag {
+ case tar.TypeDir:
+ os.MkdirAll(target, 0700) //nolint:errcheck
+ case tar.TypeReg:
+ f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
+ if err != nil {
+ return err
+ }
+ if _, err := io.Copy(f, tr); err != nil {
+ f.Close()
+ return err
+ }
+ f.Close()
+ }
+ }
+ slog.Info("Store restored from S3", "key", s3Key, "bytes", len(data))
+ return nil
+}
+
+// uploadStore tarballs the crypto store directory and uploads it to S3 for
+// the next run to restore.
+func uploadStore(ctx context.Context, storeDir, s3Key string) error {
+ var buf bytes.Buffer
+ gw := gzip.NewWriter(&buf)
+ tw := tar.NewWriter(gw)
+ if err := addDirToTar(tw, storeDir, "."); err != nil {
+ return err
+ }
+ if err := tw.Close(); err != nil {
+ return fmt.Errorf("finalize tar: %w", err)
+ }
+ if err := gw.Close(); err != nil {
+ return fmt.Errorf("finalize gzip: %w", err)
+ }
+ data := buf.Bytes()
+ if err := s3Put(ctx, s3Key, data, "application/gzip"); err != nil {
+ return err
+ }
+ slog.Info("Store saved to S3", "key", s3Key, "bytes", len(data))
+ return nil
+}
+
+func addDirToTar(tw *tar.Writer, baseDir, arcBase string) error {
+ entries, err := os.ReadDir(baseDir)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return nil
+ }
+ return err
+ }
+ for _, e := range entries {
+ srcPath := baseDir + "/" + e.Name()
+ arcPath := arcBase + "/" + e.Name()
+ if e.IsDir() {
+ _ = tw.WriteHeader(&tar.Header{
+ Typeflag: tar.TypeDir,
+ Name: arcPath + "/",
+ Mode: 0700,
+ })
+ if err := addDirToTar(tw, srcPath, arcPath); err != nil {
+ return err
+ }
+ continue
+ }
+ info, err := e.Info()
+ if err != nil {
+ continue
+ }
+ f, err := os.Open(srcPath)
+ if err != nil {
+ continue
+ }
+ if err := tw.WriteHeader(&tar.Header{
+ Typeflag: tar.TypeReg,
+ Name: arcPath,
+ Size: info.Size(),
+ Mode: 0600,
+ }); err != nil {
+ f.Close()
+ return err
+ }
+ _, copyErr := io.Copy(tw, f)
+ f.Close()
+ if copyErr != nil {
+ return copyErr
+ }
+ }
+ return nil
+}
