commit 74e21978e9ca4c8e15d5cab42346cdadd174c742 · edavis.dev/monarch

+1 -1

cmd/monarch/backfill.go

···

       11
        
       		RelayHost:             "https://bsky.network",

     

       12
        
       	}

     

       13
        
       

     

       14
       -
       	return backfill.NewBackfiller("backfiller", store, h.HandleUpsert, h.HandleUpsert, h.HandleDelete, opts)

     

       15
        
       }

···

       11
        
       		RelayHost:             "https://bsky.network",

     

       12
        
       	}

     

       13
        
       

     

       14
       +
       	return backfill.NewBackfiller("backfiller", store, h.HandleCreate, h.HandleUpdate, h.HandleDelete, opts)

     

       15
        
       }

+46 -8

cmd/monarch/handlers.go

···

       11
        
       	"tangled.sh/edavis.dev/monarch/models"

     

       12
        
       )

     

       13
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       14
        
       type HandlerService struct {

     

       15
        
       	store *gorm.DB

     

       16
        
       }

     
···

       65
        
       	}

     

       66
        
       }

     

       67
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       68
        
       // handles both creates and updates

     

       69
       -
       func (hs *HandlerService) HandleUpsert(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       70
        
       	uri, err := syntax.ParseATURI(fmt.Sprintf("at://%s/%s", repo, path))

     

       71
        
       	if err != nil {

     

       72
        
       		return fmt.Errorf("error parsing at-uri: %w", err)

     

       73
        
       	}

     

       0
        
       
     

       0
        
       
     

       74
        
       

     

       75
        
       	switch uri.Collection() {

     

       76
        
       	case syntax.NSID("app.bsky.actor.profile"):

     
···

       116
        
       			return nil

     

       117
        
       		}

     

       118
        
       

     

       119
       -
       		db := hs.store.Session(&gorm.Session{DryRun: true})

     

       120
       -
       		repost := models.NewFeedRepost(uri, *rec)

     

       121
       -
       		// stmt := db.Where(models.FeedRepost{ID: string(uri)}).Assign(repost).FirstOrCreate(&models.FeedRepost{}).Statement

     

       122
       -
       		stmt := db.Create(repost).Statement

     

       123
       -
       

     

       124
       -
       		slog.Info("repost sql statement", "sql", stmt.SQL.String(), "vars", stmt.Vars)

     

       125
       -
       

     

       126
        
       		// if err := hs.store.Where(models.FeedRepost{ID: string(uri)}).Assign(repost).FirstOrCreate(&models.FeedRepost{}).Error; err != nil {

     

       127
        
       		// 	return fmt.Errorf("error upserting feed repost: %w", err)

     

       128
        
       		// }

     

       0
        
       
     

       0
        
       
     

       129
        
       

     

       130
        
       	case syntax.NSID("app.bsky.feed.threadgate"):

     

       131
        
       		threadgate := models.NewFeedThreadgate(uri, *rec)

     
···

       189
        
       	}

     

       190
        
       

     

       191
        
       	return nil

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       192
        
       }

     

       193
        
       

     

       194
        
       func (hs *HandlerService) HandleDelete(ctx context.Context, repo string, rev string, path string) error {

···

       11
        
       	"tangled.sh/edavis.dev/monarch/models"

     

       12
        
       )

     

       13
        
       

     

       14
       +
       type Action int

     

       15
       +
       

     

       16
       +
       const (

     

       17
       +
       	ActionCreate Action = iota

     

       18
       +
       	ActionUpdate

     

       19
       +
       )

     

       20
       +
       

     

       21
        
       type HandlerService struct {

     

       22
        
       	store *gorm.DB

     

       23
        
       }

     
···

       72
        
       	}

     

       73
        
       }

     

       74
        
       

     

       75
       +
       type helper struct {

     

       76
       +
       	db *gorm.DB

     

       77
       +
       	uri syntax.ATURI

     

       78
       +
       	record []byte

     

       79
       +
       	action Action

     

       80
       +
       }

     

       81
       +
       

     

       82
       +
       func NewHelper(db *gorm.DB, uri syntax.ATURI, record []byte, action Action) *helper {

     

       83
       +
       	return &helper{db, uri, record, action}

     

       84
       +
       }

     

       85
       +
       

     

       86
       +
       func upsertRecord[T any](maker func(syntax.ATURI, []byte) T, opts *helper) error {

     

       87
       +
       	obj := maker(opts.uri, opts.record)

     

       88
       +
       	switch opts.action {

     

       89
       +
       	case ActionCreate:

     

       90
       +
       		if err := opts.db.Create(obj).Error; err != nil {

     

       91
       +
       			return fmt.Errorf("error inserting %s record: %w", opts.uri.Collection(), err)

     

       92
       +
       		}

     

       93
       +
       	case ActionUpdate:

     

       94
       +
       		if err := opts.db.Save(obj).Error; err != nil {

     

       95
       +
       			return fmt.Errorf("error updating %s record: %w", opts.uri.Collection(), err)

     

       96
       +
       		}

     

       97
       +
       	}

     

       98
       +
       	return nil

     

       99
       +
       }

     

       100
       +
       

     

       101
        
       // handles both creates and updates

     

       102
       +
       func (hs *HandlerService) HandleUpsert(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid, action Action) error {

     

       103
        
       	uri, err := syntax.ParseATURI(fmt.Sprintf("at://%s/%s", repo, path))

     

       104
        
       	if err != nil {

     

       105
        
       		return fmt.Errorf("error parsing at-uri: %w", err)

     

       106
        
       	}

     

       107
       +
       

     

       108
       +
       	opts := NewHelper(hs.store, uri, *rec, action)

     

       109
        
       

     

       110
        
       	switch uri.Collection() {

     

       111
        
       	case syntax.NSID("app.bsky.actor.profile"):

     
···

       151
        
       			return nil

     

       152
        
       		}

     

       153
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       154
        
       		// if err := hs.store.Where(models.FeedRepost{ID: string(uri)}).Assign(repost).FirstOrCreate(&models.FeedRepost{}).Error; err != nil {

     

       155
        
       		// 	return fmt.Errorf("error upserting feed repost: %w", err)

     

       156
        
       		// }

     

       157
       +
       

     

       158
       +
       		return upsertRecord(models.NewFeedRepost, opts)

     

       159
        
       

     

       160
        
       	case syntax.NSID("app.bsky.feed.threadgate"):

     

       161
        
       		threadgate := models.NewFeedThreadgate(uri, *rec)

     
···

       219
        
       	}

     

       220
        
       

     

       221
        
       	return nil

     

       222
       +
       }

     

       223
       +
       

     

       224
       +
       func (hs *HandlerService) HandleCreate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       225
       +
       	return hs.HandleUpsert(ctx, repo, rev, path, rec, cid, ActionCreate)

     

       226
       +
       }

     

       227
       +
       

     

       228
       +
       func (hs *HandlerService) HandleUpdate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       229
       +
       	return hs.HandleUpsert(ctx, repo, rev, path, rec, cid, ActionUpdate)

     

       230
        
       }

     

       231
        
       

     

       232
        
       func (hs *HandlerService) HandleDelete(ctx context.Context, repo string, rev string, path string) error {