+1

internal/firehose/config.go

··· 22 22 atproto.NSIDGrinder, 23 23 atproto.NSIDBrewer, 24 24 atproto.NSIDLike, 25 25 + atproto.NSIDComment, 25 26 } 26 27 27 28 // Config holds configuration for the Jetstream consumer

+4

internal/firehose/consumer.go

··· 349 349 if err := c.index.UpsertLike(event.DID, commit.RKey, subjectURI); err != nil { 350 350 log.Warn().Err(err).Str("did", event.DID).Str("subject", subjectURI).Msg("failed to index like") 351 351 } 352 352 + // Create notification for the like 353 353 + c.index.CreateLikeNotification(event.DID, subjectURI) 352 354 } 353 355 } 354 356 } ··· 379 381 if err := c.index.UpsertComment(event.DID, commit.RKey, subjectURI, parentURI, commit.CID, text, createdAt); err != nil { 380 382 log.Warn().Err(err).Str("did", event.DID).Str("subject", subjectURI).Msg("failed to index comment") 381 383 } 384 384 + // Create notification for the comment 385 385 + c.index.CreateCommentNotification(event.DID, subjectURI, parentURI) 382 386 } 383 387 } 384 388 }

+22 -1

internal/firehose/index.go

··· 174 174 BucketCommentCounts, 175 175 BucketCommentsByActor, 176 176 BucketCommentChildren, 177 177 + BucketNotifications, 178 178 + BucketNotificationsMeta, 177 179 } 178 180 for _, bucket := range buckets { 179 181 if _, err := tx.CreateBucketIfNotExists(bucket); err != nil { ··· 1310 1312 existingSubject := commentsByActor.Get(actorKey) 1311 1313 isNew := existingSubject == nil 1312 1314 1315 1315 + // If the comment already exists, delete the old entry from BucketComments 1316 1316 + // to prevent duplicates (the key includes timestamp which may differ between calls) 1317 1317 + if !isNew { 1318 1318 + oldPrefix := []byte(string(existingSubject) + ":") 1319 1319 + suffix := ":" + actorDID + ":" + rkey 1320 1320 + cur := comments.Cursor() 1321 1321 + for k, _ := cur.Seek(oldPrefix); k != nil && strings.HasPrefix(string(k), string(oldPrefix)); k, _ = cur.Next() { 1322 1322 + if strings.HasSuffix(string(k), suffix) { 1323 1323 + _ = comments.Delete(k) 1324 1324 + break 1325 1325 + } 1326 1326 + } 1327 1327 + } 1328 1328 + 1313 1329 // Extract parent rkey from parent URI if present 1314 1330 var parentRKey string 1315 1331 if parentURI != "" { ··· 1382 1398 1383 1399 actorKey := []byte(actorDID + ":" + rkey) 1384 1400 1385 1385 - // Check if comment exists 1401 1401 + // Check if comment exists and get subject URI from index 1386 1402 existingSubject := commentsByActor.Get(actorKey) 1387 1403 if existingSubject == nil { 1388 1404 return nil // Comment doesn't exist, nothing to do 1405 1405 + } 1406 1406 + 1407 1407 + // Use the subject URI from the index if not provided 1408 1408 + if subjectURI == "" { 1409 1409 + subjectURI = string(existingSubject) 1389 1410 } 1390 1411 1391 1412 // Find and delete the comment by iterating over comments with matching subject

+257

internal/firehose/notifications.go

··· 1 1 + package firehose 2 2 + 3 3 + import ( 4 4 + "encoding/json" 5 5 + "fmt" 6 6 + "strings" 7 7 + "time" 8 8 + 9 9 + "arabica/internal/models" 10 10 + 11 11 + "github.com/rs/zerolog/log" 12 12 + bolt "go.etcd.io/bbolt" 13 13 + ) 14 14 + 15 15 + // Bucket names for notifications 16 16 + var ( 17 17 + // BucketNotifications stores notifications: {target_did}:{inverted_timestamp}:{id} -> {Notification JSON} 18 18 + BucketNotifications = []byte("notifications") 19 19 + 20 20 + // BucketNotificationsMeta stores per-user metadata: {target_did}:last_read -> {timestamp RFC3339} 21 21 + BucketNotificationsMeta = []byte("notifications_meta") 22 22 + ) 23 23 + 24 24 + // CreateNotification stores a notification for the target user. 25 25 + // Deduplicates by (type + actorDID + subjectURI) to prevent duplicates from backfills. 26 26 + // Self-notifications (actorDID == targetDID) are silently skipped. 27 27 + func (idx *FeedIndex) CreateNotification(targetDID string, notif models.Notification) error { 28 28 + if targetDID == "" || targetDID == notif.ActorDID { 29 29 + return nil // skip self-notifications 30 30 + } 31 31 + 32 32 + return idx.db.Update(func(tx *bolt.Tx) error { 33 33 + b := tx.Bucket(BucketNotifications) 34 34 + 35 35 + // Deduplication: scan for existing notification with same type+actor+subject 36 36 + prefix := []byte(targetDID + ":") 37 37 + c := b.Cursor() 38 38 + for k, v := c.Seek(prefix); k != nil && strings.HasPrefix(string(k), string(prefix)); k, v = c.Next() { 39 39 + var existing models.Notification 40 40 + if err := json.Unmarshal(v, &existing); err != nil { 41 41 + continue 42 42 + } 43 43 + if existing.Type == notif.Type && existing.ActorDID == notif.ActorDID && existing.SubjectURI == notif.SubjectURI { 44 44 + return nil // duplicate, skip 45 45 + } 46 46 + } 47 47 + 48 48 + // Generate ID from timestamp 49 49 + if notif.ID == "" { 50 50 + notif.ID = fmt.Sprintf("%d", notif.CreatedAt.UnixNano()) 51 51 + } 52 52 + 53 53 + data, err := json.Marshal(notif) 54 54 + if err != nil { 55 55 + return fmt.Errorf("failed to marshal notification: %w", err) 56 56 + } 57 57 + 58 58 + // Key: {target_did}:{inverted_timestamp}:{id} for reverse chronological order 59 59 + inverted := ^uint64(notif.CreatedAt.UnixNano()) 60 60 + key := fmt.Sprintf("%s:%016x:%s", targetDID, inverted, notif.ID) 61 61 + return b.Put([]byte(key), data) 62 62 + }) 63 63 + } 64 64 + 65 65 + // GetNotifications returns notifications for a user, newest first. 66 66 + // Uses cursor-based pagination. Returns notifications, next cursor, and error. 67 67 + func (idx *FeedIndex) GetNotifications(targetDID string, limit int, cursor string) ([]models.Notification, string, error) { 68 68 + var notifications []models.Notification 69 69 + var nextCursor string 70 70 + 71 71 + if limit <= 0 { 72 72 + limit = 20 73 73 + } 74 74 + 75 75 + // Get last_read timestamp for marking read status 76 76 + lastRead := idx.getLastRead(targetDID) 77 77 + 78 78 + err := idx.db.View(func(tx *bolt.Tx) error { 79 79 + b := tx.Bucket(BucketNotifications) 80 80 + c := b.Cursor() 81 81 + 82 82 + prefix := []byte(targetDID + ":") 83 83 + var k, v []byte 84 84 + 85 85 + if cursor != "" { 86 86 + // Seek to cursor position, then advance past it 87 87 + k, v = c.Seek([]byte(cursor)) 88 88 + if k != nil && string(k) == cursor { 89 89 + k, v = c.Next() 90 90 + } 91 91 + } else { 92 92 + k, v = c.Seek(prefix) 93 93 + } 94 94 + 95 95 + var lastKey []byte 96 96 + count := 0 97 97 + for ; k != nil && strings.HasPrefix(string(k), string(prefix)); k, v = c.Next() { 98 98 + if count >= limit { 99 99 + // There are more items beyond our limit 100 100 + nextCursor = string(lastKey) 101 101 + break 102 102 + } 103 103 + 104 104 + var notif models.Notification 105 105 + if err := json.Unmarshal(v, &notif); err != nil { 106 106 + continue 107 107 + } 108 108 + 109 109 + // Determine read status based on last_read timestamp 110 110 + if !lastRead.IsZero() && !notif.CreatedAt.After(lastRead) { 111 111 + notif.Read = true 112 112 + } 113 113 + 114 114 + notifications = append(notifications, notif) 115 115 + lastKey = make([]byte, len(k)) 116 116 + copy(lastKey, k) 117 117 + count++ 118 118 + } 119 119 + 120 120 + return nil 121 121 + }) 122 122 + 123 123 + return notifications, nextCursor, err 124 124 + } 125 125 + 126 126 + // GetUnreadCount returns the number of unread notifications for a user. 127 127 + func (idx *FeedIndex) GetUnreadCount(targetDID string) int { 128 128 + if targetDID == "" { 129 129 + return 0 130 130 + } 131 131 + 132 132 + lastRead := idx.getLastRead(targetDID) 133 133 + 134 134 + var count int 135 135 + _ = idx.db.View(func(tx *bolt.Tx) error { 136 136 + b := tx.Bucket(BucketNotifications) 137 137 + c := b.Cursor() 138 138 + 139 139 + prefix := []byte(targetDID + ":") 140 140 + for k, v := c.Seek(prefix); k != nil && strings.HasPrefix(string(k), string(prefix)); k, v = c.Next() { 141 141 + var notif models.Notification 142 142 + if err := json.Unmarshal(v, &notif); err != nil { 143 143 + continue 144 144 + } 145 145 + // If no last_read set, all are unread 146 146 + if lastRead.IsZero() || notif.CreatedAt.After(lastRead) { 147 147 + count++ 148 148 + } else { 149 149 + // Since keys are in reverse chronological order, 150 150 + // once we hit a read notification, all remaining are also read 151 151 + break 152 152 + } 153 153 + } 154 154 + return nil 155 155 + }) 156 156 + 157 157 + return count 158 158 + } 159 159 + 160 160 + // MarkAllRead updates the last_read timestamp to now for the user. 161 161 + func (idx *FeedIndex) MarkAllRead(targetDID string) error { 162 162 + return idx.db.Update(func(tx *bolt.Tx) error { 163 163 + b := tx.Bucket(BucketNotificationsMeta) 164 164 + key := []byte(targetDID + ":last_read") 165 165 + return b.Put(key, []byte(time.Now().Format(time.RFC3339Nano))) 166 166 + }) 167 167 + } 168 168 + 169 169 + // getLastRead returns the last_read timestamp for a user. 170 170 + func (idx *FeedIndex) getLastRead(targetDID string) time.Time { 171 171 + var lastRead time.Time 172 172 + _ = idx.db.View(func(tx *bolt.Tx) error { 173 173 + b := tx.Bucket(BucketNotificationsMeta) 174 174 + v := b.Get([]byte(targetDID + ":last_read")) 175 175 + if v != nil { 176 176 + if t, err := time.Parse(time.RFC3339Nano, string(v)); err == nil { 177 177 + lastRead = t 178 178 + } 179 179 + } 180 180 + return nil 181 181 + }) 182 182 + return lastRead 183 183 + } 184 184 + 185 185 + // parseTargetDID extracts the DID from an AT-URI (at://did:plc:xxx/collection/rkey) 186 186 + func parseTargetDID(atURI string) string { 187 187 + if !strings.HasPrefix(atURI, "at://") { 188 188 + return "" 189 189 + } 190 190 + rest := atURI[5:] // strip "at://" 191 191 + parts := strings.SplitN(rest, "/", 2) 192 192 + if len(parts) == 0 { 193 193 + return "" 194 194 + } 195 195 + did := parts[0] 196 196 + if !strings.HasPrefix(did, "did:") { 197 197 + return "" 198 198 + } 199 199 + return did 200 200 + } 201 201 + 202 202 + // CreateLikeNotification creates a notification for a like event 203 203 + func (idx *FeedIndex) CreateLikeNotification(actorDID, subjectURI string) { 204 204 + targetDID := parseTargetDID(subjectURI) 205 205 + if targetDID == "" || targetDID == actorDID { 206 206 + return 207 207 + } 208 208 + 209 209 + notif := models.Notification{ 210 210 + Type: models.NotificationLike, 211 211 + ActorDID: actorDID, 212 212 + SubjectURI: subjectURI, 213 213 + CreatedAt: time.Now(), 214 214 + } 215 215 + 216 216 + if err := idx.CreateNotification(targetDID, notif); err != nil { 217 217 + log.Warn().Err(err).Str("actor", actorDID).Str("subject", subjectURI).Msg("failed to create like notification") 218 218 + } 219 219 + } 220 220 + 221 221 + // CreateCommentNotification creates notifications for a comment event. 222 222 + // Notifies the brew owner (comment) and the parent comment author (reply). 223 223 + func (idx *FeedIndex) CreateCommentNotification(actorDID, subjectURI, parentURI string) { 224 224 + now := time.Now() 225 225 + 226 226 + // Notify the brew owner 227 227 + targetDID := parseTargetDID(subjectURI) 228 228 + if targetDID != "" && targetDID != actorDID { 229 229 + notif := models.Notification{ 230 230 + Type: models.NotificationComment, 231 231 + ActorDID: actorDID, 232 232 + SubjectURI: subjectURI, 233 233 + CreatedAt: now, 234 234 + } 235 235 + if err := idx.CreateNotification(targetDID, notif); err != nil { 236 236 + log.Warn().Err(err).Str("actor", actorDID).Str("subject", subjectURI).Msg("failed to create comment notification") 237 237 + } 238 238 + } 239 239 + 240 240 + // If this is a reply, also notify the parent comment's author. 241 241 + // We store the brew's subjectURI (not the parent comment URI) so the 242 242 + // notification links directly to the brew page with comments. 243 243 + if parentURI != "" { 244 244 + parentAuthorDID := parseTargetDID(parentURI) 245 245 + if parentAuthorDID != "" && parentAuthorDID != actorDID && parentAuthorDID != targetDID { 246 246 + replyNotif := models.Notification{ 247 247 + Type: models.NotificationCommentReply, 248 248 + ActorDID: actorDID, 249 249 + SubjectURI: subjectURI, // brew URI, not parent comment URI 250 250 + CreatedAt: now, 251 251 + } 252 252 + if err := idx.CreateNotification(parentAuthorDID, replyNotif); err != nil { 253 253 + log.Warn().Err(err).Str("actor", actorDID).Str("parent", parentURI).Msg("failed to create reply notification") 254 254 + } 255 255 + } 256 256 + } 257 257 + }

+281

internal/firehose/notifications_test.go

··· 1 1 + package firehose 2 2 + 3 3 + import ( 4 4 + "os" 5 5 + "path/filepath" 6 6 + "testing" 7 7 + "time" 8 8 + 9 9 + "arabica/internal/models" 10 10 + 11 11 + "github.com/stretchr/testify/assert" 12 12 + ) 13 13 + 14 14 + func newTestIndex(t *testing.T) *FeedIndex { 15 15 + t.Helper() 16 16 + dir := t.TempDir() 17 17 + path := filepath.Join(dir, "test-index.db") 18 18 + idx, err := NewFeedIndex(path, time.Hour) 19 19 + assert.NoError(t, err) 20 20 + t.Cleanup(func() { 21 21 + idx.Close() 22 22 + os.Remove(path) 23 23 + }) 24 24 + return idx 25 25 + } 26 26 + 27 27 + func TestCreateNotification(t *testing.T) { 28 28 + idx := newTestIndex(t) 29 29 + 30 30 + targetDID := "did:plc:target123" 31 31 + actorDID := "did:plc:actor456" 32 32 + subjectURI := "at://did:plc:target123/social.arabica.alpha.brew/abc" 33 33 + 34 34 + notif := models.Notification{ 35 35 + Type: models.NotificationLike, 36 36 + ActorDID: actorDID, 37 37 + SubjectURI: subjectURI, 38 38 + CreatedAt: time.Now(), 39 39 + } 40 40 + 41 41 + err := idx.CreateNotification(targetDID, notif) 42 42 + assert.NoError(t, err) 43 43 + 44 44 + // Verify it was created 45 45 + notifications, _, err := idx.GetNotifications(targetDID, 10, "") 46 46 + assert.NoError(t, err) 47 47 + assert.Len(t, notifications, 1) 48 48 + assert.Equal(t, models.NotificationLike, notifications[0].Type) 49 49 + assert.Equal(t, actorDID, notifications[0].ActorDID) 50 50 + assert.Equal(t, subjectURI, notifications[0].SubjectURI) 51 51 + } 52 52 + 53 53 + func TestCreateNotification_SkipsSelfNotification(t *testing.T) { 54 54 + idx := newTestIndex(t) 55 55 + 56 56 + selfDID := "did:plc:self123" 57 57 + 58 58 + notif := models.Notification{ 59 59 + Type: models.NotificationLike, 60 60 + ActorDID: selfDID, 61 61 + SubjectURI: "at://did:plc:self123/social.arabica.alpha.brew/abc", 62 62 + CreatedAt: time.Now(), 63 63 + } 64 64 + 65 65 + err := idx.CreateNotification(selfDID, notif) 66 66 + assert.NoError(t, err) 67 67 + 68 68 + notifications, _, err := idx.GetNotifications(selfDID, 10, "") 69 69 + assert.NoError(t, err) 70 70 + assert.Empty(t, notifications) 71 71 + } 72 72 + 73 73 + func TestCreateNotification_Deduplication(t *testing.T) { 74 74 + idx := newTestIndex(t) 75 75 + 76 76 + targetDID := "did:plc:target123" 77 77 + notif := models.Notification{ 78 78 + Type: models.NotificationLike, 79 79 + ActorDID: "did:plc:actor456", 80 80 + SubjectURI: "at://did:plc:target123/social.arabica.alpha.brew/abc", 81 81 + CreatedAt: time.Now(), 82 82 + } 83 83 + 84 84 + // Create the same notification twice 85 85 + assert.NoError(t, idx.CreateNotification(targetDID, notif)) 86 86 + assert.NoError(t, idx.CreateNotification(targetDID, notif)) 87 87 + 88 88 + notifications, _, err := idx.GetNotifications(targetDID, 10, "") 89 89 + assert.NoError(t, err) 90 90 + assert.Len(t, notifications, 1) // should be deduplicated 91 91 + } 92 92 + 93 93 + func TestGetUnreadCount(t *testing.T) { 94 94 + idx := newTestIndex(t) 95 95 + 96 96 + targetDID := "did:plc:target123" 97 97 + baseTime := time.Now().Add(-time.Minute) 98 98 + 99 99 + // Initially zero 100 100 + assert.Equal(t, 0, idx.GetUnreadCount(targetDID)) 101 101 + 102 102 + // Add some notifications 103 103 + for i := 0; i < 3; i++ { 104 104 + notif := models.Notification{ 105 105 + Type: models.NotificationLike, 106 106 + ActorDID: "did:plc:actor" + string(rune('a'+i)), 107 107 + SubjectURI: "at://did:plc:target123/social.arabica.alpha.brew/abc", 108 108 + CreatedAt: baseTime.Add(time.Duration(i) * time.Second), 109 109 + } 110 110 + assert.NoError(t, idx.CreateNotification(targetDID, notif)) 111 111 + } 112 112 + 113 113 + assert.Equal(t, 3, idx.GetUnreadCount(targetDID)) 114 114 + } 115 115 + 116 116 + func TestMarkAllRead(t *testing.T) { 117 117 + idx := newTestIndex(t) 118 118 + 119 119 + targetDID := "did:plc:target123" 120 120 + baseTime := time.Now().Add(-time.Minute) // use past times to avoid race 121 121 + 122 122 + // Add notifications 123 123 + for i := 0; i < 3; i++ { 124 124 + notif := models.Notification{ 125 125 + Type: models.NotificationLike, 126 126 + ActorDID: "did:plc:actor" + string(rune('a'+i)), 127 127 + SubjectURI: "at://did:plc:target123/social.arabica.alpha.brew/abc", 128 128 + CreatedAt: baseTime.Add(time.Duration(i) * time.Second), 129 129 + } 130 130 + assert.NoError(t, idx.CreateNotification(targetDID, notif)) 131 131 + } 132 132 + 133 133 + assert.Equal(t, 3, idx.GetUnreadCount(targetDID)) 134 134 + 135 135 + // Mark all as read 136 136 + assert.NoError(t, idx.MarkAllRead(targetDID)) 137 137 + assert.Equal(t, 0, idx.GetUnreadCount(targetDID)) 138 138 + 139 139 + // Notifications still exist, but are marked as read 140 140 + notifications, _, err := idx.GetNotifications(targetDID, 10, "") 141 141 + assert.NoError(t, err) 142 142 + assert.Len(t, notifications, 3) 143 143 + for _, n := range notifications { 144 144 + assert.True(t, n.Read) 145 145 + } 146 146 + } 147 147 + 148 148 + func TestGetNotifications_Pagination(t *testing.T) { 149 149 + idx := newTestIndex(t) 150 150 + 151 151 + targetDID := "did:plc:target123" 152 152 + baseTime := time.Now().Add(-time.Minute) 153 153 + 154 154 + // Add 5 notifications 155 155 + for i := 0; i < 5; i++ { 156 156 + notif := models.Notification{ 157 157 + Type: models.NotificationLike, 158 158 + ActorDID: "did:plc:actor" + string(rune('a'+i)), 159 159 + SubjectURI: "at://did:plc:target123/social.arabica.alpha.brew/abc", 160 160 + CreatedAt: baseTime.Add(time.Duration(i) * time.Second), 161 161 + } 162 162 + assert.NoError(t, idx.CreateNotification(targetDID, notif)) 163 163 + } 164 164 + 165 165 + // Get first page of 3 166 166 + page1, cursor1, err := idx.GetNotifications(targetDID, 3, "") 167 167 + assert.NoError(t, err) 168 168 + assert.Len(t, page1, 3) 169 169 + assert.NotEmpty(t, cursor1) 170 170 + 171 171 + // Get second page 172 172 + page2, cursor2, err := idx.GetNotifications(targetDID, 3, cursor1) 173 173 + assert.NoError(t, err) 174 174 + assert.Len(t, page2, 2) 175 175 + assert.Empty(t, cursor2) 176 176 + } 177 177 + 178 178 + func TestParseTargetDID(t *testing.T) { 179 179 + tests := []struct { 180 180 + name string 181 181 + uri string 182 182 + expected string 183 183 + }{ 184 184 + { 185 185 + name: "valid brew URI", 186 186 + uri: "at://did:plc:abc123/social.arabica.alpha.brew/xyz", 187 187 + expected: "did:plc:abc123", 188 188 + }, 189 189 + { 190 190 + name: "valid like URI", 191 191 + uri: "at://did:web:example.com/social.arabica.alpha.like/xyz", 192 192 + expected: "did:web:example.com", 193 193 + }, 194 194 + { 195 195 + name: "empty string", 196 196 + uri: "", 197 197 + expected: "", 198 198 + }, 199 199 + { 200 200 + name: "not an AT URI", 201 201 + uri: "https://example.com/something", 202 202 + expected: "", 203 203 + }, 204 204 + { 205 205 + name: "AT URI without did prefix", 206 206 + uri: "at://handle.example.com/collection/rkey", 207 207 + expected: "", 208 208 + }, 209 209 + } 210 210 + 211 211 + for _, tt := range tests { 212 212 + t.Run(tt.name, func(t *testing.T) { 213 213 + result := parseTargetDID(tt.uri) 214 214 + assert.Equal(t, tt.expected, result) 215 215 + }) 216 216 + } 217 217 + } 218 218 + 219 219 + func TestCreateLikeNotification(t *testing.T) { 220 220 + idx := newTestIndex(t) 221 221 + 222 222 + actorDID := "did:plc:actor456" 223 223 + subjectURI := "at://did:plc:target123/social.arabica.alpha.brew/abc" 224 224 + 225 225 + idx.CreateLikeNotification(actorDID, subjectURI) 226 226 + 227 227 + notifications, _, err := idx.GetNotifications("did:plc:target123", 10, "") 228 228 + assert.NoError(t, err) 229 229 + assert.Len(t, notifications, 1) 230 230 + assert.Equal(t, models.NotificationLike, notifications[0].Type) 231 231 + } 232 232 + 233 233 + func TestCreateLikeNotification_SkipsSelf(t *testing.T) { 234 234 + idx := newTestIndex(t) 235 235 + 236 236 + selfDID := "did:plc:target123" 237 237 + subjectURI := "at://did:plc:target123/social.arabica.alpha.brew/abc" 238 238 + 239 239 + idx.CreateLikeNotification(selfDID, subjectURI) 240 240 + 241 241 + notifications, _, err := idx.GetNotifications("did:plc:target123", 10, "") 242 242 + assert.NoError(t, err) 243 243 + assert.Empty(t, notifications) 244 244 + } 245 245 + 246 246 + func TestCreateCommentNotification(t *testing.T) { 247 247 + idx := newTestIndex(t) 248 248 + 249 249 + actorDID := "did:plc:actor456" 250 250 + subjectURI := "at://did:plc:target123/social.arabica.alpha.brew/abc" 251 251 + 252 252 + idx.CreateCommentNotification(actorDID, subjectURI, "") 253 253 + 254 254 + notifications, _, err := idx.GetNotifications("did:plc:target123", 10, "") 255 255 + assert.NoError(t, err) 256 256 + assert.Len(t, notifications, 1) 257 257 + assert.Equal(t, models.NotificationComment, notifications[0].Type) 258 258 + } 259 259 + 260 260 + func TestCreateCommentNotification_WithReply(t *testing.T) { 261 261 + idx := newTestIndex(t) 262 262 + 263 263 + actorDID := "did:plc:actor456" 264 264 + subjectURI := "at://did:plc:brewowner/social.arabica.alpha.brew/abc" 265 265 + parentURI := "at://did:plc:commenter789/social.arabica.alpha.comment/xyz" 266 266 + 267 267 + idx.CreateCommentNotification(actorDID, subjectURI, parentURI) 268 268 + 269 269 + // Brew owner gets a comment notification 270 270 + brewOwnerNotifs, _, err := idx.GetNotifications("did:plc:brewowner", 10, "") 271 271 + assert.NoError(t, err) 272 272 + assert.Len(t, brewOwnerNotifs, 1) 273 273 + assert.Equal(t, models.NotificationComment, brewOwnerNotifs[0].Type) 274 274 + 275 275 + // Parent comment author gets a reply notification with the brew URI (not the parent comment URI) 276 276 + commenterNotifs, _, err := idx.GetNotifications("did:plc:commenter789", 10, "") 277 277 + assert.NoError(t, err) 278 278 + assert.Len(t, commenterNotifs, 1) 279 279 + assert.Equal(t, models.NotificationCommentReply, commenterNotifs[0].Type) 280 280 + assert.Equal(t, subjectURI, commenterNotifs[0].SubjectURI) 281 281 + }

+18 -30

internal/handlers/handlers.go

··· 248 248 isModerator = h.moderationService.IsModerator(didStr) 249 249 } 250 250 251 251 + // Get unread notification count for authenticated users 252 252 + var unreadNotifCount int 253 253 + if h.feedIndex != nil && didStr != "" { 254 254 + unreadNotifCount = h.feedIndex.GetUnreadCount(didStr) 255 255 + } 256 256 + 251 257 return &components.LayoutData{ 252 252 - Title: title, 253 253 - IsAuthenticated: isAuthenticated, 254 254 - UserDID: didStr, 255 255 - UserProfile: userProfile, 256 256 - CSPNonce: middleware.CSPNonceFromContext(r.Context()), 257 257 - IsModerator: isModerator, 258 258 + Title: title, 259 259 + IsAuthenticated: isAuthenticated, 260 260 + UserDID: didStr, 261 261 + UserProfile: userProfile, 262 262 + CSPNonce: middleware.CSPNonceFromContext(r.Context()), 263 263 + IsModerator: isModerator, 264 264 + UnreadNotificationCount: unreadNotifCount, 258 265 } 259 266 } 260 267 ··· 328 335 if err := h.feedIndex.UpsertComment(didStr, comment.RKey, subjectURI, parentURI, comment.CID, text, comment.CreatedAt); err != nil { 329 336 log.Warn().Err(err).Str("did", didStr).Str("rkey", comment.RKey).Str("subject_uri", subjectURI).Msg("Failed to upsert comment in feed index") 330 337 } 338 338 + // Create notification for the comment/reply 339 339 + h.feedIndex.CreateCommentNotification(didStr, subjectURI, parentURI) 331 340 } 332 341 333 342 // Return the updated comment section with threaded comments ··· 373 382 return 374 383 } 375 384 376 376 - // Get the comment to find its subject URI before deletion 377 377 - comments, err := store.ListUserComments(r.Context()) 378 378 - if err != nil { 379 379 - http.Error(w, "Failed to get comments", http.StatusInternalServerError) 380 380 - log.Error().Err(err).Msg("Failed to get user comments") 381 381 - return 382 382 - } 383 383 - 384 384 - var subjectURI string 385 385 - for _, c := range comments { 386 386 - if c.RKey == rkey { 387 387 - subjectURI = c.SubjectURI 388 388 - break 389 389 - } 390 390 - } 391 391 - 392 392 - if subjectURI == "" { 393 393 - http.Error(w, "Comment not found", http.StatusNotFound) 394 394 - return 395 395 - } 396 396 - 397 397 - // Delete the comment 385 385 + // Delete the comment from the user's PDS 398 386 if err := store.DeleteCommentByRKey(r.Context(), rkey); err != nil { 399 387 http.Error(w, "Failed to delete comment", http.StatusInternalServerError) 400 388 log.Error().Err(err).Msg("Failed to delete comment") ··· 405 393 406 394 // Update firehose index 407 395 if h.feedIndex != nil { 408 408 - if err := h.feedIndex.DeleteComment(didStr, rkey, subjectURI); err != nil { 409 409 - log.Warn().Err(err).Str("did", didStr).Str("rkey", rkey).Str("subject_uri", subjectURI).Msg("Failed to delete comment from feed index") 396 396 + if err := h.feedIndex.DeleteComment(didStr, rkey, ""); err != nil { 397 397 + log.Warn().Err(err).Str("did", didStr).Str("rkey", rkey).Msg("Failed to delete comment from feed index") 410 398 } 411 399 } 412 400

+115

internal/handlers/notifications.go

··· 1 1 + package handlers 2 2 + 3 3 + import ( 4 4 + "fmt" 5 5 + "net/http" 6 6 + "strings" 7 7 + 8 8 + "arabica/internal/atproto" 9 9 + "arabica/internal/web/pages" 10 10 + 11 11 + "github.com/rs/zerolog/log" 12 12 + ) 13 13 + 14 14 + // HandleNotifications renders the notifications page 15 15 + func (h *Handler) HandleNotifications(w http.ResponseWriter, r *http.Request) { 16 16 + layoutData, didStr, isAuthenticated := h.layoutDataFromRequest(r, "Notifications") 17 17 + if !isAuthenticated { 18 18 + http.Redirect(w, r, "/login", http.StatusSeeOther) 19 19 + return 20 20 + } 21 21 + 22 22 + cursor := r.URL.Query().Get("cursor") 23 23 + 24 24 + var props pages.NotificationsProps 25 25 + 26 26 + if h.feedIndex != nil { 27 27 + notifications, nextCursor, err := h.feedIndex.GetNotifications(didStr, 30, cursor) 28 28 + if err != nil { 29 29 + log.Error().Err(err).Str("did", didStr).Msg("Failed to get notifications") 30 30 + http.Error(w, "Failed to load notifications", http.StatusInternalServerError) 31 31 + return 32 32 + } 33 33 + 34 34 + props.NextCursor = nextCursor 35 35 + 36 36 + // Resolve actor profiles and links for each notification 37 37 + for _, notif := range notifications { 38 38 + item := pages.NotificationItem{ 39 39 + Notification: notif, 40 40 + Link: resolveNotificationLink(notif.SubjectURI), 41 41 + } 42 42 + 43 43 + profile, err := h.feedIndex.GetProfile(r.Context(), notif.ActorDID) 44 44 + if err == nil && profile != nil { 45 45 + item.ActorHandle = profile.Handle 46 46 + if profile.DisplayName != nil { 47 47 + item.ActorDisplayName = *profile.DisplayName 48 48 + } 49 49 + if profile.Avatar != nil { 50 50 + item.ActorAvatar = *profile.Avatar 51 51 + } 52 52 + } else { 53 53 + item.ActorHandle = notif.ActorDID 54 54 + } 55 55 + 56 56 + props.Notifications = append(props.Notifications, item) 57 57 + } 58 58 + 59 59 + // Mark all as read when the page is viewed 60 60 + if err := h.feedIndex.MarkAllRead(didStr); err != nil { 61 61 + log.Warn().Err(err).Str("did", didStr).Msg("Failed to mark notifications as read on view") 62 62 + } 63 63 + } 64 64 + 65 65 + if err := pages.Notifications(layoutData, props).Render(r.Context(), w); err != nil { 66 66 + http.Error(w, "Failed to render page", http.StatusInternalServerError) 67 67 + log.Error().Err(err).Msg("Failed to render notifications page") 68 68 + } 69 69 + } 70 70 + 71 71 + // HandleNotificationsMarkRead marks all notifications as read 72 72 + func (h *Handler) HandleNotificationsMarkRead(w http.ResponseWriter, r *http.Request) { 73 73 + didStr, err := atproto.GetAuthenticatedDID(r.Context()) 74 74 + if err != nil || didStr == "" { 75 75 + http.Error(w, "Authentication required", http.StatusUnauthorized) 76 76 + return 77 77 + } 78 78 + 79 79 + if h.feedIndex != nil { 80 80 + if err := h.feedIndex.MarkAllRead(didStr); err != nil { 81 81 + log.Error().Err(err).Str("did", didStr).Msg("Failed to mark notifications as read") 82 82 + http.Error(w, "Failed to mark notifications as read", http.StatusInternalServerError) 83 83 + return 84 84 + } 85 85 + } 86 86 + 87 87 + // Redirect back to notifications page 88 88 + http.Redirect(w, r, "/notifications", http.StatusSeeOther) 89 89 + } 90 90 + 91 91 + // resolveNotificationLink converts a SubjectURI (AT-URI) to a local page URL. 92 92 + // All notification types store a brew AT-URI as SubjectURI. 93 93 + // Format: at://did:plc:xxx/social.arabica.alpha.brew/rkey -> /brews/rkey?owner=did:plc:xxx 94 94 + func resolveNotificationLink(subjectURI string) string { 95 95 + if !strings.HasPrefix(subjectURI, "at://") { 96 96 + return "" 97 97 + } 98 98 + 99 99 + rest := subjectURI[5:] // strip "at://" 100 100 + parts := strings.SplitN(rest, "/", 3) 101 101 + if len(parts) < 3 { 102 102 + return "" 103 103 + } 104 104 + 105 105 + did := parts[0] 106 106 + collection := parts[1] 107 107 + rkey := parts[2] 108 108 + 109 109 + switch collection { 110 110 + case atproto.NSIDBrew: 111 111 + return fmt.Sprintf("/brews/%s?owner=%s", rkey, did) 112 112 + default: 113 113 + return "" 114 114 + } 115 115 + }

+10 -9

internal/middleware/logging.go

··· 141 141 } 142 142 143 143 func getCookies(r *http.Request) string { 144 144 - sb := strings.Builder{} 145 145 - loggedCookies := []string{"account_did", "session_id"} 146 146 - for _, name := range loggedCookies { 147 147 - // TODO: check if `c.Domain == "arabica.social"` if we start adding it 148 148 - if c, err := r.Cookie(name); err == nil { 149 149 - _, _ = sb.WriteString(c.Value + "; ") 150 150 - } 151 151 - } 152 152 - return sb.String() 144 144 + // sb := strings.Builder{} 145 145 + // loggedCookies := []string{"account_did", "session_id"} 146 146 + // for _, name := range loggedCookies { 147 147 + // // TODO: check if `c.Domain == "arabica.social"` if we start adding it 148 148 + // if c, err := r.Cookie(name); err == nil { 149 149 + // _, _ = sb.WriteString(c.Name + "=" + c.Value + "; ") 150 150 + // } 151 151 + // } 152 152 + // return sb.String() 153 153 + return r.Header.Get("cookies") 153 154 }

+19

internal/models/models.go

··· 413 413 return nil 414 414 } 415 415 416 416 + // NotificationType represents the type of notification 417 417 + type NotificationType string 418 418 + 419 419 + const ( 420 420 + NotificationLike NotificationType = "like" 421 421 + NotificationComment NotificationType = "comment" 422 422 + NotificationCommentReply NotificationType = "comment_reply" 423 423 + ) 424 424 + 425 425 + // Notification represents a notification for a user 426 426 + type Notification struct { 427 427 + ID string `json:"id"` // Unique key (timestamp-based) 428 428 + Type NotificationType `json:"type"` // like, comment, comment_reply 429 429 + ActorDID string `json:"actor_did"` // Who performed the action 430 430 + SubjectURI string `json:"subject_uri"` // The brew/comment that was acted on 431 431 + CreatedAt time.Time `json:"created_at"` 432 432 + Read bool `json:"read"` 433 433 + } 434 434 + 416 435 // Report represents a user-submitted content report 417 436 // TODO: Store reports in database (BoltDB or SQLite) for moderation review 418 437 type Report struct {

+4

internal/routing/routing.go

··· 107 107 mux.HandleFunc("GET /api/modals/roaster/new", h.HandleRoasterModalNew) 108 108 mux.HandleFunc("GET /api/modals/roaster/{id}", h.HandleRoasterModalEdit) 109 109 110 110 + // Notification routes 111 111 + mux.HandleFunc("GET /notifications", h.HandleNotifications) 112 112 + mux.Handle("POST /api/notifications/read", cop.Handler(http.HandlerFunc(h.HandleNotificationsMarkRead))) 113 113 + 110 114 // Profile routes (public user profiles) 111 115 mux.HandleFunc("GET /profile/{actor}", h.HandleProfile) 112 116

+28 -5

internal/web/components/header.templ

··· 1 1 package components 2 2 3 3 - import "arabica/internal/web/bff" 3 3 + import ( 4 4 + "fmt" 5 5 + "arabica/internal/web/bff" 6 6 + ) 4 7 5 8 // HeaderProps contains all properties for the header component 6 9 type HeaderProps struct { 7 7 - IsAuthenticated bool 8 8 - UserProfile *bff.UserProfile 9 9 - UserDID string 10 10 - IsModerator bool // Show admin link in dropdown 10 10 + IsAuthenticated bool 11 11 + UserProfile *bff.UserProfile 12 12 + UserDID string 13 13 + IsModerator bool // Show admin link in dropdown 14 14 + UnreadNotificationCount int // Badge count for bell icon 11 15 } 12 16 13 17 templ Header(isAuthenticated bool, userProfile *bff.UserProfile, userDID string) { ··· 30 34 <!-- Navigation links --> 31 35 <div class="flex items-center gap-4"> 32 36 if props.IsAuthenticated { 37 37 + <!-- Notification bell --> 38 38 + <a href="/notifications" class="relative hover:opacity-80 transition p-1" title="Notifications"> 39 39 + <svg class="w-6 h-6" fill="none" stroke="currentColor" stroke-width="1.5" viewBox="0 0 24 24"> 40 40 + <path stroke-linecap="round" stroke-linejoin="round" d="M14.857 17.082a23.848 23.848 0 0 0 5.454-1.31A8.967 8.967 0 0 1 18 9.75V9A6 6 0 0 0 6 9v.75a8.967 8.967 0 0 1-2.312 6.022c1.733.64 3.56 1.085 5.455 1.31m5.714 0a24.255 24.255 0 0 1-5.714 0m5.714 0a3 3 0 1 1-5.714 0"></path> 41 41 + </svg> 42 42 + if props.UnreadNotificationCount > 0 { 43 43 + <span class="absolute -top-1 -right-1 bg-amber-400 text-brown-900 text-xs font-bold rounded-full min-w-[18px] h-[18px] flex items-center justify-center px-1 shadow-sm"> 44 44 + { formatNotificationCount(props.UnreadNotificationCount) } 45 45 + </span> 46 46 + } 47 47 + </a> 33 48 <!-- User profile dropdown --> 34 49 <div x-data="{ open: false }" class="relative"> 35 50 <button @click="open = !open" @click.outside="open = false" class="flex items-center gap-2 hover:opacity-80 transition focus:outline-none"> ··· 122 137 } 123 138 return "" 124 139 } 140 140 + 141 141 + // formatNotificationCount formats the notification badge count (99+ for large numbers) 142 142 + func formatNotificationCount(count int) string { 143 143 + if count > 99 { 144 144 + return "99+" 145 145 + } 146 146 + return fmt.Sprintf("%d", count) 147 147 + }

+7 -5

internal/web/components/layout.templ

··· 9 9 UserDID string 10 10 UserProfile *bff.UserProfile 11 11 CSPNonce string 12 12 - IsModerator bool // User has moderation permissions 12 12 + IsModerator bool // User has moderation permissions 13 13 + UnreadNotificationCount int // Number of unread notifications 13 14 14 15 // OpenGraph metadata (optional, uses defaults if empty) 15 16 OGTitle string // Falls back to Title + " - Arabica" ··· 133 134 } 134 135 > 135 136 @HeaderWithProps(HeaderProps{ 136 136 - IsAuthenticated: data.IsAuthenticated, 137 137 - UserProfile: data.UserProfile, 138 138 - UserDID: data.UserDID, 139 139 - IsModerator: data.IsModerator, 137 137 + IsAuthenticated: data.IsAuthenticated, 138 138 + UserProfile: data.UserProfile, 139 139 + UserDID: data.UserDID, 140 140 + IsModerator: data.IsModerator, 141 141 + UnreadNotificationCount: data.UnreadNotificationCount, 140 142 }) 141 143 <main class="flex-grow container mx-auto py-8" data-transition> 142 144 @content

+133

internal/web/pages/notifications.templ

··· 1 1 + package pages 2 2 + 3 3 + import ( 4 4 + "arabica/internal/models" 5 5 + "arabica/internal/web/bff" 6 6 + "arabica/internal/web/components" 7 7 + ) 8 8 + 9 9 + // NotificationsProps holds the data for the notifications page 10 10 + type NotificationsProps struct { 11 11 + Notifications []NotificationItem 12 12 + NextCursor string 13 13 + } 14 14 + 15 15 + // NotificationItem is a notification with resolved profile data 16 16 + type NotificationItem struct { 17 17 + models.Notification 18 18 + ActorHandle string 19 19 + ActorDisplayName string 20 20 + ActorAvatar string 21 21 + Link string // Resolved local URL (e.g. /brews/rkey?owner=did) 22 22 + } 23 23 + 24 24 + templ Notifications(layout *components.LayoutData, props NotificationsProps) { 25 25 + @components.Layout(layout, notificationsContent(props)) 26 26 + } 27 27 + 28 28 + templ notificationsContent(props NotificationsProps) { 29 29 + <div class="page-container-md"> 30 30 + <div class="flex items-center justify-between mb-8"> 31 31 + <div class="flex items-center gap-3"> 32 32 + @components.BackButton() 33 33 + <h1 class="text-3xl font-bold text-brown-900">Notifications</h1> 34 34 + </div> 35 35 + if len(props.Notifications) > 0 { 36 36 + <form method="POST" action="/api/notifications/read"> 37 37 + <button type="submit" class="btn-secondary text-sm"> 38 38 + Mark all as read 39 39 + </button> 40 40 + </form> 41 41 + } 42 42 + </div> 43 43 + if len(props.Notifications) == 0 { 44 44 + @components.EmptyState(components.EmptyStateProps{ 45 45 + Message: "No notifications yet", 46 46 + SubMessage: "When someone likes or comments on your brews, you'll see it here.", 47 47 + }) 48 48 + } else { 49 49 + <div class="space-y-2"> 50 50 + for _, notif := range props.Notifications { 51 51 + @notificationRow(notif) 52 52 + } 53 53 + </div> 54 54 + if props.NextCursor != "" { 55 55 + <div class="mt-6 text-center"> 56 56 + <a 57 57 + href={ templ.SafeURL("/notifications?cursor=" + props.NextCursor) } 58 58 + class="btn-secondary" 59 59 + > 60 60 + Load more 61 61 + </a> 62 62 + </div> 63 63 + } 64 64 + } 65 65 + </div> 66 66 + } 67 67 + 68 68 + templ notificationRow(notif NotificationItem) { 69 69 + <a 70 70 + href={ templ.SafeURL(notifClickURL(notif)) } 71 71 + class={ "card card-inner flex items-start gap-3 p-4 no-underline hover:bg-brown-50 transition-colors cursor-pointer", templ.KV("bg-amber-50/50 border-amber-200", !notif.Read) } 72 72 + > 73 73 + <!-- Actor avatar --> 74 74 + <div class="shrink-0"> 75 75 + @components.Avatar(components.AvatarProps{ 76 76 + AvatarURL: notif.ActorAvatar, 77 77 + DisplayName: notifActorName(notif), 78 78 + Size: "sm", 79 79 + }) 80 80 + </div> 81 81 + <!-- Content --> 82 82 + <div class="flex-1 min-w-0"> 83 83 + <p class="text-sm text-brown-800"> 84 84 + <span class="font-semibold text-brown-900"> 85 85 + { notifActorName(notif) } 86 86 + </span> 87 87 + { " " } 88 88 + { notifActionText(notif.Type) } 89 89 + </p> 90 90 + <p class="text-xs text-brown-400 mt-1"> 91 91 + { bff.FormatTimeAgo(notif.CreatedAt) } 92 92 + </p> 93 93 + </div> 94 94 + <!-- Unread indicator --> 95 95 + if !notif.Read { 96 96 + <div class="shrink-0 mt-1"> 97 97 + <span class="block w-2 h-2 rounded-full bg-amber-400"></span> 98 98 + </div> 99 99 + } 100 100 + </a> 101 101 + } 102 102 + 103 103 + func notifActorName(notif NotificationItem) string { 104 104 + if notif.ActorDisplayName != "" { 105 105 + return notif.ActorDisplayName 106 106 + } 107 107 + if notif.ActorHandle != "" { 108 108 + return notif.ActorHandle 109 109 + } 110 110 + return notif.ActorDID 111 111 + } 112 112 + 113 113 + func notifActionText(t models.NotificationType) string { 114 114 + switch t { 115 115 + case models.NotificationLike: 116 116 + return "liked your brew" 117 117 + case models.NotificationComment: 118 118 + return "commented on your brew" 119 119 + case models.NotificationCommentReply: 120 120 + return "replied to your comment" 121 121 + default: 122 122 + return "interacted with your content" 123 123 + } 124 124 + } 125 125 + 126 126 + // notifClickURL returns the URL to navigate to when clicking a notification. 127 127 + // Uses the pre-resolved Link field, falling back to the notifications page. 128 128 + func notifClickURL(notif NotificationItem) string { 129 129 + if notif.Link != "" { 130 130 + return notif.Link 131 131 + } 132 132 + return "/notifications" 133 133 + }