events.go (5292B)
1 // events.go — per-event processing: media download, profile-change recording, 2 // and the historyEvent record type shared by both pagination paths. 3 package main 4 5 import ( 6 "context" 7 "encoding/json" 8 "log/slog" 9 "strings" 10 11 "maunium.net/go/mautrix" 12 "maunium.net/go/mautrix/event" 13 "maunium.net/go/mautrix/id" 14 ) 15 16 // historyEvent is the on-disk representation of a single room event stored 17 // in JSONL files under prefix/history/<safeKey>/<run>.jsonl.age. 18 type historyEvent struct { 19 EventID string `json:"event_id"` 20 Sender string `json:"sender"` 21 Type string `json:"type"` 22 Timestamp int64 `json:"origin_server_ts"` 23 Content json.RawMessage `json:"content"` 24 Encrypted bool `json:"encrypted,omitempty"` 25 } 26 27 func eventToRecord(ev *event.Event, wasEncrypted bool) historyEvent { 28 return historyEvent{ 29 EventID: string(ev.ID), 30 Sender: string(ev.Sender), 31 Type: ev.Type.Type, 32 Timestamp: ev.Timestamp, 33 Content: ev.Content.VeryRaw, 34 Encrypted: wasEncrypted, 35 } 36 } 37 38 // roomSafeKey converts a room ID into a string safe for use as an S3 key 39 // component by replacing '/' and ':' with '_'. 40 func roomSafeKey(roomID id.RoomID) string { 41 return strings.NewReplacer("/", "_", ":", "_").Replace(string(roomID)) 42 } 43 44 // processEvent downloads media and records profile changes for a single event. 45 // The event must already be decrypted before this is called. 46 func processEvent(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string, isDM bool) { 47 if ev.Type == event.EventEncrypted { 48 return 49 } 50 51 if ev.Type == event.EventMessage { 52 var content struct { 53 MsgType string `json:"msgtype"` 54 URL string `json:"url"` 55 File json.RawMessage `json:"file"` 56 } 57 if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil { 58 mediaURL := content.URL 59 if mediaURL == "" && content.File != nil { 60 var f struct { 61 URL string `json:"url"` 62 } 63 if json.Unmarshal(content.File, &f) == nil { 64 mediaURL = f.URL 65 } 66 } 67 if mediaMsgTypes[content.MsgType] && mediaURL != "" { 68 if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) { 69 if err := downloadAndStoreMedia(ctx, client, mediaURL, prefix, "media"); err != nil { 70 slog.Warn("Media download failed", "url", mediaURL, "error", err) 71 } 72 } 73 } 74 } 75 } 76 77 if ev.Type == event.EventSticker { 78 var content struct { 79 URL string `json:"url"` 80 File json.RawMessage `json:"file"` 81 } 82 if err := json.Unmarshal(ev.Content.VeryRaw, &content); err == nil { 83 mediaURL := content.URL 84 if mediaURL == "" && content.File != nil { 85 var f struct { 86 URL string `json:"url"` 87 } 88 if json.Unmarshal(content.File, &f) == nil { 89 mediaURL = f.URL 90 } 91 } 92 if mediaURL != "" { 93 if isDM || strings.HasSuffix(string(ev.Sender), ":"+ourServerName) { 94 if err := downloadAndStoreMedia(ctx, client, mediaURL, prefix, "media"); err != nil { 95 slog.Warn("Sticker download failed", "url", mediaURL, "error", err) 96 } 97 } 98 } 99 } 100 } 101 102 if ev.Type == event.StateMember { 103 recordProfileUpdate(ctx, client, ev, roomID, prefix) 104 } 105 } 106 107 type profileUpdateRecord struct { 108 TS int64 `json:"ts"` 109 EventID string `json:"event_id"` 110 UserID string `json:"user_id"` 111 RoomID string `json:"room_id"` 112 DisplayNameOld string `json:"displayname_old,omitempty"` 113 DisplayNameNew string `json:"displayname_new,omitempty"` 114 AvatarOld string `json:"avatar_old,omitempty"` 115 AvatarNew string `json:"avatar_new,omitempty"` 116 } 117 118 // recordProfileUpdate appends a JSONL entry to S3 when a member event shows a 119 // display-name or avatar change for a local user; also downloads the new avatar. 120 func recordProfileUpdate(ctx context.Context, client *mautrix.Client, ev *event.Event, roomID id.RoomID, prefix string) { 121 if ev.StateKey == nil { 122 return 123 } 124 stateKey := *ev.StateKey 125 if !strings.HasSuffix(stateKey, ":"+ourServerName) { 126 return 127 } 128 var content, prevContent struct { 129 Displayname string `json:"displayname"` 130 AvatarURL string `json:"avatar_url"` 131 } 132 if err := json.Unmarshal(ev.Content.VeryRaw, &content); err != nil { 133 return 134 } 135 if ev.Unsigned.PrevContent != nil { 136 _ = json.Unmarshal(ev.Unsigned.PrevContent.VeryRaw, &prevContent) 137 } 138 if content.Displayname == prevContent.Displayname && content.AvatarURL == prevContent.AvatarURL { 139 return 140 } 141 rec := profileUpdateRecord{ 142 TS: ev.Timestamp, 143 EventID: string(ev.ID), 144 UserID: stateKey, 145 RoomID: string(roomID), 146 } 147 if content.Displayname != prevContent.Displayname { 148 rec.DisplayNameOld = prevContent.Displayname 149 rec.DisplayNameNew = content.Displayname 150 } 151 if content.AvatarURL != prevContent.AvatarURL { 152 rec.AvatarOld = prevContent.AvatarURL 153 rec.AvatarNew = content.AvatarURL 154 if content.AvatarURL != "" { 155 if err := downloadAndStoreMedia(ctx, client, content.AvatarURL, prefix, "avatars"); err != nil { 156 slog.Warn("Avatar download failed", "url", content.AvatarURL, "error", err) 157 } 158 } 159 } 160 line, err := json.Marshal(rec) 161 if err != nil { 162 return 163 } 164 s3Key := prefix + "/profile-updates.jsonl" 165 existing, _ := s3Get(ctx, s3Key) 166 _ = s3Put(ctx, s3Key, append(existing, append(line, '\n')...), "application/x-ndjson") 167 }