commit e4608b6b16b04bbd0efeee3ce9b45c973dcdc0b3 · bretton.dev/coves

+243

internal/atproto/jetstream/post_consumer.go

···

       1
       +
       package jetstream

     

       2
       +
       

     

       3
       +
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
       +
       	"Coves/internal/core/posts"

     

       6
       +
       	"Coves/internal/core/users"

     

       7
       +
       	"context"

     

       8
       +
       	"encoding/json"

     

       9
       +
       	"fmt"

     

       10
       +
       	"log"

     

       11
       +
       	"strings"

     

       12
       +
       	"time"

     

       13
       +
       )

     

       14
       +
       

     

       15
       +
       // PostEventConsumer consumes post-related events from Jetstream

     

       16
       +
       // Currently handles only CREATE operations for social.coves.post.record

     

       17
       +
       // UPDATE and DELETE handlers will be added when those features are implemented

     

       18
       +
       type PostEventConsumer struct {

     

       19
       +
       	postRepo      posts.Repository

     

       20
       +
       	communityRepo communities.Repository

     

       21
       +
       	userService   users.UserService

     

       22
       +
       }

     

       23
       +
       

     

       24
       +
       // NewPostEventConsumer creates a new Jetstream consumer for post events

     

       25
       +
       func NewPostEventConsumer(

     

       26
       +
       	postRepo posts.Repository,

     

       27
       +
       	communityRepo communities.Repository,

     

       28
       +
       	userService users.UserService,

     

       29
       +
       ) *PostEventConsumer {

     

       30
       +
       	return &PostEventConsumer{

     

       31
       +
       		postRepo:      postRepo,

     

       32
       +
       		communityRepo: communityRepo,

     

       33
       +
       		userService:   userService,

     

       34
       +
       	}

     

       35
       +
       }

     

       36
       +
       

     

       37
       +
       // HandleEvent processes a Jetstream event for post records

     

       38
       +
       // Currently only handles CREATE operations - UPDATE/DELETE deferred until those features exist

     

       39
       +
       func (c *PostEventConsumer) HandleEvent(ctx context.Context, event *JetstreamEvent) error {

     

       40
       +
       	// We only care about commit events for post records

     

       41
       +
       	if event.Kind != "commit" || event.Commit == nil {

     

       42
       +
       		return nil

     

       43
       +
       	}

     

       44
       +
       

     

       45
       +
       	commit := event.Commit

     

       46
       +
       

     

       47
       +
       	// Only handle post record creation for now

     

       48
       +
       	// UPDATE and DELETE will be added when we implement those features

     

       49
       +
       	if commit.Collection == "social.coves.post.record" && commit.Operation == "create" {

     

       50
       +
       		return c.createPost(ctx, event.Did, commit)

     

       51
       +
       	}

     

       52
       +
       

     

       53
       +
       	// Silently ignore other operations (update, delete) and other collections

     

       54
       +
       	return nil

     

       55
       +
       }

     

       56
       +
       

     

       57
       +
       // createPost indexes a new post from the firehose

     

       58
       +
       func (c *PostEventConsumer) createPost(ctx context.Context, repoDID string, commit *CommitEvent) error {

     

       59
       +
       	if commit.Record == nil {

     

       60
       +
       		return fmt.Errorf("post create event missing record data")

     

       61
       +
       	}

     

       62
       +
       

     

       63
       +
       	// Parse the post record

     

       64
       +
       	postRecord, err := parsePostRecord(commit.Record)

     

       65
       +
       	if err != nil {

     

       66
       +
       		return fmt.Errorf("failed to parse post record: %w", err)

     

       67
       +
       	}

     

       68
       +
       

     

       69
       +
       	// SECURITY: Validate this is a legitimate post event

     

       70
       +
       	if err := c.validatePostEvent(ctx, repoDID, postRecord); err != nil {

     

       71
       +
       		log.Printf("🚨 SECURITY: Rejecting post event: %v", err)

     

       72
       +
       		return err

     

       73
       +
       	}

     

       74
       +
       

     

       75
       +
       	// Build AT-URI for this post

     

       76
       +
       	// Format: at://community_did/social.coves.post.record/rkey

     

       77
       +
       	uri := fmt.Sprintf("at://%s/social.coves.post.record/%s", repoDID, commit.RKey)

     

       78
       +
       

     

       79
       +
       	// Parse timestamp from record

     

       80
       +
       	createdAt, err := time.Parse(time.RFC3339, postRecord.CreatedAt)

     

       81
       +
       	if err != nil {

     

       82
       +
       		// Fallback to current time if parsing fails

     

       83
       +
       		log.Printf("Warning: Failed to parse createdAt timestamp, using current time: %v", err)

     

       84
       +
       		createdAt = time.Now()

     

       85
       +
       	}

     

       86
       +
       

     

       87
       +
       	// Build post entity

     

       88
       +
       	post := &posts.Post{

     

       89
       +
       		URI:          uri,

     

       90
       +
       		CID:          commit.CID,

     

       91
       +
       		RKey:         commit.RKey,

     

       92
       +
       		AuthorDID:    postRecord.Author,

     

       93
       +
       		CommunityDID: postRecord.Community,

     

       94
       +
       		Title:        postRecord.Title,

     

       95
       +
       		Content:      postRecord.Content,

     

       96
       +
       		CreatedAt:    createdAt,

     

       97
       +
       		IndexedAt:    time.Now(),

     

       98
       +
       		// Stats remain at 0 (no votes yet)

     

       99
       +
       		UpvoteCount:   0,

     

       100
       +
       		DownvoteCount: 0,

     

       101
       +
       		Score:         0,

     

       102
       +
       		CommentCount:  0,

     

       103
       +
       	}

     

       104
       +
       

     

       105
       +
       	// Serialize JSON fields (facets, embed, labels)

     

       106
       +
       	if postRecord.Facets != nil {

     

       107
       +
       		facetsJSON, marshalErr := json.Marshal(postRecord.Facets)

     

       108
       +
       		if marshalErr == nil {

     

       109
       +
       			facetsStr := string(facetsJSON)

     

       110
       +
       			post.ContentFacets = &facetsStr

     

       111
       +
       		}

     

       112
       +
       	}

     

       113
       +
       

     

       114
       +
       	if postRecord.Embed != nil {

     

       115
       +
       		embedJSON, marshalErr := json.Marshal(postRecord.Embed)

     

       116
       +
       		if marshalErr == nil {

     

       117
       +
       			embedStr := string(embedJSON)

     

       118
       +
       			post.Embed = &embedStr

     

       119
       +
       		}

     

       120
       +
       	}

     

       121
       +
       

     

       122
       +
       	if len(postRecord.ContentLabels) > 0 {

     

       123
       +
       		labelsJSON, marshalErr := json.Marshal(postRecord.ContentLabels)

     

       124
       +
       		if marshalErr == nil {

     

       125
       +
       			labelsStr := string(labelsJSON)

     

       126
       +
       			post.ContentLabels = &labelsStr

     

       127
       +
       		}

     

       128
       +
       	}

     

       129
       +
       

     

       130
       +
       	// Index in AppView database (idempotent - safe for Jetstream replays)

     

       131
       +
       	err = c.postRepo.Create(ctx, post)

     

       132
       +
       	if err != nil {

     

       133
       +
       		// Check if it already exists (idempotency)

     

       134
       +
       		if posts.IsConflict(err) {

     

       135
       +
       			log.Printf("Post already indexed: %s", uri)

     

       136
       +
       			return nil

     

       137
       +
       		}

     

       138
       +
       		return fmt.Errorf("failed to index post: %w", err)

     

       139
       +
       	}

     

       140
       +
       

     

       141
       +
       	log.Printf("✓ Indexed post: %s (author: %s, community: %s, rkey: %s)",

     

       142
       +
       		uri, post.AuthorDID, post.CommunityDID, commit.RKey)

     

       143
       +
       	return nil

     

       144
       +
       }

     

       145
       +
       

     

       146
       +
       // validatePostEvent performs security validation on post events

     

       147
       +
       // This prevents malicious actors from indexing fake posts

     

       148
       +
       func (c *PostEventConsumer) validatePostEvent(ctx context.Context, repoDID string, post *PostRecordFromJetstream) error {

     

       149
       +
       	// CRITICAL SECURITY CHECK:

     

       150
       +
       	// Posts MUST come from community repositories, not user repositories

     

       151
       +
       	// This prevents users from creating posts that appear to be from communities they don't control

     

       152
       +
       	//

     

       153
       +
       	// Example attack prevented:

     

       154
       +
       	//   - User creates post in their own repo (at://user_did/social.coves.post.record/xyz)

     

       155
       +
       	//   - Claims it's for community X (community field = community_did)

     

       156
       +
       	//   - Without this check, fake post would be indexed

     

       157
       +
       	//

     

       158
       +
       	// With this check:

     

       159
       +
       	//   - We verify event.Did (repo owner) == post.community (claimed community)

     

       160
       +
       	//   - Reject if mismatch

     

       161
       +
       	if repoDID != post.Community {

     

       162
       +
       		return fmt.Errorf("repository DID (%s) doesn't match community DID (%s) - posts must come from community repos",

     

       163
       +
       			repoDID, post.Community)

     

       164
       +
       	}

     

       165
       +
       

     

       166
       +
       	// CRITICAL: Verify community exists in AppView

     

       167
       +
       	// Posts MUST reference valid communities (enforced by FK constraint)

     

       168
       +
       	// If community isn't indexed yet, we must reject the post

     

       169
       +
       	// Jetstream will replay events, so the post will be indexed once community is ready

     

       170
       +
       	_, err := c.communityRepo.GetByDID(ctx, post.Community)

     

       171
       +
       	if err != nil {

     

       172
       +
       		if communities.IsNotFound(err) {

     

       173
       +
       			// Reject - community must be indexed before posts

     

       174
       +
       			// This maintains referential integrity and prevents orphaned posts

     

       175
       +
       			return fmt.Errorf("community not found: %s - cannot index post before community", post.Community)

     

       176
       +
       		}

     

       177
       +
       		// Database error or other issue

     

       178
       +
       		return fmt.Errorf("failed to verify community exists: %w", err)

     

       179
       +
       	}

     

       180
       +
       

     

       181
       +
       	// CRITICAL: Verify author exists in AppView

     

       182
       +
       	// Every post MUST have a valid author (enforced by FK constraint)

     

       183
       +
       	// Even though posts live in community repos, they belong to specific authors

     

       184
       +
       	// If author isn't indexed yet, we must reject the post

     

       185
       +
       	_, err = c.userService.GetUserByDID(ctx, post.Author)

     

       186
       +
       	if err != nil {

     

       187
       +
       		// Check if it's a "not found" error using string matching

     

       188
       +
       		// (users package doesn't export IsNotFound)

     

       189
       +
       		if err.Error() == "user not found" || strings.Contains(err.Error(), "not found") {

     

       190
       +
       			// Reject - author must be indexed before posts

     

       191
       +
       			// This maintains referential integrity and prevents orphaned posts

     

       192
       +
       			return fmt.Errorf("author not found: %s - cannot index post before author", post.Author)

     

       193
       +
       		}

     

       194
       +
       		// Database error or other issue

     

       195
       +
       		return fmt.Errorf("failed to verify author exists: %w", err)

     

       196
       +
       	}

     

       197
       +
       

     

       198
       +
       	return nil

     

       199
       +
       }

     

       200
       +
       

     

       201
       +
       // PostRecordFromJetstream represents a post record as received from Jetstream

     

       202
       +
       // Matches the structure written to PDS via social.coves.post.record

     

       203
       +
       type PostRecordFromJetstream struct {

     

       204
       +
       	OriginalAuthor interface{}            `json:"originalAuthor,omitempty"`

     

       205
       +
       	FederatedFrom  interface{}            `json:"federatedFrom,omitempty"`

     

       206
       +
       	Location       interface{}            `json:"location,omitempty"`

     

       207
       +
       	Title          *string                `json:"title,omitempty"`

     

       208
       +
       	Content        *string                `json:"content,omitempty"`

     

       209
       +
       	Embed          map[string]interface{} `json:"embed,omitempty"`

     

       210
       +
       	Type           string                 `json:"$type"`

     

       211
       +
       	Community      string                 `json:"community"`

     

       212
       +
       	Author         string                 `json:"author"`

     

       213
       +
       	CreatedAt      string                 `json:"createdAt"`

     

       214
       +
       	Facets         []interface{}          `json:"facets,omitempty"`

     

       215
       +
       	ContentLabels  []string               `json:"contentLabels,omitempty"`

     

       216
       +
       }

     

       217
       +
       

     

       218
       +
       // parsePostRecord converts a raw Jetstream record map to a PostRecordFromJetstream

     

       219
       +
       func parsePostRecord(record map[string]interface{}) (*PostRecordFromJetstream, error) {

     

       220
       +
       	// Marshal to JSON and back to ensure proper type conversion

     

       221
       +
       	recordJSON, err := json.Marshal(record)

     

       222
       +
       	if err != nil {

     

       223
       +
       		return nil, fmt.Errorf("failed to marshal record: %w", err)

     

       224
       +
       	}

     

       225
       +
       

     

       226
       +
       	var post PostRecordFromJetstream

     

       227
       +
       	if err := json.Unmarshal(recordJSON, &post); err != nil {

     

       228
       +
       		return nil, fmt.Errorf("failed to unmarshal post record: %w", err)

     

       229
       +
       	}

     

       230
       +
       

     

       231
       +
       	// Validate required fields

     

       232
       +
       	if post.Community == "" {

     

       233
       +
       		return nil, fmt.Errorf("post record missing community field")

     

       234
       +
       	}

     

       235
       +
       	if post.Author == "" {

     

       236
       +
       		return nil, fmt.Errorf("post record missing author field")

     

       237
       +
       	}

     

       238
       +
       	if post.CreatedAt == "" {

     

       239
       +
       		return nil, fmt.Errorf("post record missing createdAt field")

     

       240
       +
       	}

     

       241
       +
       

     

       242
       +
       	return &post, nil

     

       243
       +
       }

+125

internal/atproto/jetstream/post_jetstream_connector.go

···

       1
       +
       package jetstream

     

       2
       +
       

     

       3
       +
       import (

     

       4
       +
       	"context"

     

       5
       +
       	"encoding/json"

     

       6
       +
       	"fmt"

     

       7
       +
       	"log"

     

       8
       +
       	"sync"

     

       9
       +
       	"time"

     

       10
       +
       

     

       11
       +
       	"github.com/gorilla/websocket"

     

       12
       +
       )

     

       13
       +
       

     

       14
       +
       // PostJetstreamConnector handles WebSocket connection to Jetstream for post events

     

       15
       +
       type PostJetstreamConnector struct {

     

       16
       +
       	consumer *PostEventConsumer

     

       17
       +
       	wsURL    string

     

       18
       +
       }

     

       19
       +
       

     

       20
       +
       // NewPostJetstreamConnector creates a new Jetstream WebSocket connector for post events

     

       21
       +
       func NewPostJetstreamConnector(consumer *PostEventConsumer, wsURL string) *PostJetstreamConnector {

     

       22
       +
       	return &PostJetstreamConnector{

     

       23
       +
       		consumer: consumer,

     

       24
       +
       		wsURL:    wsURL,

     

       25
       +
       	}

     

       26
       +
       }

     

       27
       +
       

     

       28
       +
       // Start begins consuming events from Jetstream

     

       29
       +
       // Runs indefinitely, reconnecting on errors

     

       30
       +
       func (c *PostJetstreamConnector) Start(ctx context.Context) error {

     

       31
       +
       	log.Printf("Starting Jetstream post consumer: %s", c.wsURL)

     

       32
       +
       

     

       33
       +
       	for {

     

       34
       +
       		select {

     

       35
       +
       		case <-ctx.Done():

     

       36
       +
       			log.Println("Jetstream post consumer shutting down")

     

       37
       +
       			return ctx.Err()

     

       38
       +
       		default:

     

       39
       +
       			if err := c.connect(ctx); err != nil {

     

       40
       +
       				log.Printf("Jetstream post connection error: %v. Retrying in 5s...", err)

     

       41
       +
       				time.Sleep(5 * time.Second)

     

       42
       +
       				continue

     

       43
       +
       			}

     

       44
       +
       		}

     

       45
       +
       	}

     

       46
       +
       }

     

       47
       +
       

     

       48
       +
       // connect establishes WebSocket connection and processes events

     

       49
       +
       func (c *PostJetstreamConnector) connect(ctx context.Context) error {

     

       50
       +
       	conn, _, err := websocket.DefaultDialer.DialContext(ctx, c.wsURL, nil)

     

       51
       +
       	if err != nil {

     

       52
       +
       		return fmt.Errorf("failed to connect to Jetstream: %w", err)

     

       53
       +
       	}

     

       54
       +
       	defer func() {

     

       55
       +
       		if closeErr := conn.Close(); closeErr != nil {

     

       56
       +
       			log.Printf("Failed to close WebSocket connection: %v", closeErr)

     

       57
       +
       		}

     

       58
       +
       	}()

     

       59
       +
       

     

       60
       +
       	log.Println("Connected to Jetstream (post consumer)")

     

       61
       +
       

     

       62
       +
       	// Set read deadline to detect connection issues

     

       63
       +
       	if err := conn.SetReadDeadline(time.Now().Add(60 * time.Second)); err != nil {

     

       64
       +
       		log.Printf("Failed to set read deadline: %v", err)

     

       65
       +
       	}

     

       66
       +
       

     

       67
       +
       	// Set pong handler to keep connection alive

     

       68
       +
       	conn.SetPongHandler(func(string) error {

     

       69
       +
       		if err := conn.SetReadDeadline(time.Now().Add(60 * time.Second)); err != nil {

     

       70
       +
       			log.Printf("Failed to set read deadline in pong handler: %v", err)

     

       71
       +
       		}

     

       72
       +
       		return nil

     

       73
       +
       	})

     

       74
       +
       

     

       75
       +
       	// Start ping ticker

     

       76
       +
       	ticker := time.NewTicker(30 * time.Second)

     

       77
       +
       	defer ticker.Stop()

     

       78
       +
       

     

       79
       +
       	done := make(chan struct{})

     

       80
       +
       	var closeOnce sync.Once // Ensure done channel is only closed once

     

       81
       +
       

     

       82
       +
       	// Ping goroutine

     

       83
       +
       	go func() {

     

       84
       +
       		for {

     

       85
       +
       			select {

     

       86
       +
       			case <-ticker.C:

     

       87
       +
       				if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(10*time.Second)); err != nil {

     

       88
       +
       					log.Printf("Failed to send ping: %v", err)

     

       89
       +
       					closeOnce.Do(func() { close(done) })

     

       90
       +
       					return

     

       91
       +
       				}

     

       92
       +
       			case <-done:

     

       93
       +
       				return

     

       94
       +
       			}

     

       95
       +
       		}

     

       96
       +
       	}()

     

       97
       +
       

     

       98
       +
       	// Read loop

     

       99
       +
       	for {

     

       100
       +
       		select {

     

       101
       +
       		case <-done:

     

       102
       +
       			return fmt.Errorf("connection closed by ping failure")

     

       103
       +
       		default:

     

       104
       +
       		}

     

       105
       +
       

     

       106
       +
       		_, message, err := conn.ReadMessage()

     

       107
       +
       		if err != nil {

     

       108
       +
       			closeOnce.Do(func() { close(done) })

     

       109
       +
       			return fmt.Errorf("read error: %w", err)

     

       110
       +
       		}

     

       111
       +
       

     

       112
       +
       		// Parse Jetstream event

     

       113
       +
       		var event JetstreamEvent

     

       114
       +
       		if err := json.Unmarshal(message, &event); err != nil {

     

       115
       +
       			log.Printf("Failed to parse Jetstream event: %v", err)

     

       116
       +
       			continue

     

       117
       +
       		}

     

       118
       +
       

     

       119
       +
       		// Process event through consumer

     

       120
       +
       		if err := c.consumer.HandleEvent(ctx, &event); err != nil {

     

       121
       +
       			log.Printf("Failed to handle post event: %v", err)

     

       122
       +
       			// Continue processing other events even if one fails

     

       123
       +
       		}

     

       124
       +
       	}

     

       125
       +
       }