cluster

Infrastructure files for Nordgedanken and Midnightthoughts.
git clone git://archive.git.mtrnord.blog/MTRNord/cluster.git
Log | Files | Refs | README

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 }