patch of feat(consumer/ingester): use slogger · round #0 · pull #18 · yoten.app/yoten

+30 -20

internal/consumer/ingester.go

···

       4
        
       	"context"

     

       5
        
       	"encoding/json"

     

       6
        
       	"fmt"

     

       7
       -
       	"log"

     

       8
        
       	"strings"

     

       9
        
       	"time"

     

       10
        
       

     
···

       20
        
       type Ingester struct {

     

       21
        
       	Db     db.DbWrapper

     

       22
        
       	Config *config.Config

     

       0
        
       
     

       23
        
       }

     

       24
        
       

     

       25
        
       type processFunc func(ctx context.Context, e *models.Event) error

     
···

       35
        
       			}

     

       36
        
       		}()

     

       37
        
       

     

       0
        
       
     

       38
        
       		switch e.Kind {

     

       39
        
       		case models.EventKindCommit:

     

       40
        
       			switch e.Commit.Collection {

     

       41
        
       			case yoten.ActorProfileNSID:

     

       0
        
       
     

       42
        
       				err = i.ingestProfile(e)

     

       43
        
       			case yoten.FeedSessionNSID:

     

       0
        
       
     

       44
        
       				err = i.ingestStudySession(e)

     

       45
        
       			case yoten.ActivityDefNSID:

     

       0
        
       
     

       46
        
       				err = i.ingestActivityDef(e)

     

       47
        
       			case yoten.FeedResourceNSID:

     

       0
        
       
     

       48
        
       				err = i.ingestResource(e)

     

       49
        
       			case yoten.GraphFollowNSID:

     

       0
        
       
     

       50
        
       				err = i.ingestFollow(e)

     

       51
        
       			case yoten.FeedReactionNSID:

     

       0
        
       
     

       52
        
       				err = i.ingestReaction(e)

     

       53
        
       			case yoten.FeedCommentNSID:

     

       0
        
       
     

       54
        
       				err = i.ingestComment(e)

     

       55
        
       			}

     

       0
        
       
     

       56
        
       		}

     

       57
        
       		if err != nil {

     

       58
       -
       			log.Printf("failed to ingest event for collection %s: %v", e.Commit.Collection, err)

     

       59
        
       		}

     

       60
        
       

     

       61
        
       		return nil

     
···

       129
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       130
        
       		}

     

       131
        
       

     

       132
       -
       		log.Printf("upserting profile '%s' from pds request", profile.Did)

     

       133
        
       		err = db.UpsertProfile(tx, &profile)

     

       134
        
       		if err != nil {

     

       135
        
       			tx.Rollback()

     
···

       160
        
       

     

       161
        
       		date, err := time.Parse(time.RFC3339, record.Date)

     

       162
        
       		if err != nil {

     

       163
       -
       			log.Printf("invalid record: %s", err)

     

       164
        
       			return err

     

       165
        
       		}

     

       166
        
       

     
···

       227
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       228
        
       		}

     

       229
        
       

     

       230
       -
       		log.Println("upserting study session from pds request")

     

       231
        
       		err = db.UpsertStudySession(tx, &studySession, e.Commit.RKey)

     

       232
        
       		if err != nil {

     

       233
        
       			tx.Rollback()

     
···

       252
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       253
        
       		}

     

       254
        
       

     

       255
       -
       		log.Println("deleting study session from pds request")

     

       256
        
       		err = db.DeleteStudySessionByRkey(tx, did, e.Commit.RKey)

     

       257
        
       		if err != nil {

     

       258
        
       			tx.Rollback()

     
···

       344
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       345
        
       		}

     

       346
        
       

     

       347
       -
       		log.Println("upserting activity def from pds request")

     

       348
        
       		err = db.UpsertActivityDef(tx, &activityDef, e.Commit.RKey)

     

       349
        
       		if err != nil {

     

       350
        
       			tx.Rollback()

     
···

       352
        
       		}

     

       353
        
       		return tx.Commit()

     

       354
        
       	case models.CommitOperationDelete:

     

       355
       -
       		log.Println("deleting activity def from pds request")

     

       356
        
       		err = db.DeleteActivityDefByRkey(i.Db, did, e.Commit.RKey)

     

       357
        
       	}

     

       358
        
       	if err != nil {

     
···

       387
        
       

     

       388
        
       		subjectDid := record.Subject

     

       389
        
       

     

       390
       -
       		log.Println("upserting follow from pds request")

     

       391
        
       		err = db.AddFollow(tx, did, subjectDid, e.Commit.RKey)

     

       392
        
       		if err != nil {

     

       393
        
       			tx.Rollback()

     
···

       397
        
       		subjectUri := fmt.Sprintf("at://%s/%s/%s", did, yoten.GraphFollowNSID, e.Commit.RKey)

     

       398
        
       		err = db.CreateNotification(tx, subjectDid, did, subjectUri, db.NotificationTypeFollow)

     

       399
        
       		if err != nil {

     

       400
       -
       			log.Println("failed to create notification record:", err)

     

       401
        
       		}

     

       402
        
       

     

       403
        
       		return tx.Commit()

     

       404
        
       	case models.CommitOperationDelete:

     

       405
       -
       		log.Println("deleting follow from pds request")

     

       406
        
       		err = db.DeleteFollowByRkey(i.Db, did, e.Commit.RKey)

     

       407
        
       	}

     

       408
        
       	if err != nil {

     
···

       465
        
       			CreatedAt:   createdAt,

     

       466
        
       		}

     

       467
        
       

     

       468
       -
       		log.Println("upserting reaction from pds request")

     

       469
        
       		err = db.UpsertReaction(i.Db, reactionEvent)

     

       470
        
       		if err != nil {

     

       471
        
       			tx.Rollback()

     
···

       474
        
       

     

       475
        
       		err = db.CreateNotification(tx, subjectDid.String(), did, subject.String(), db.NotificationTypeReaction)

     

       476
        
       		if err != nil {

     

       477
       -
       			log.Println("failed to create notification record:", err)

     

       478
        
       		}

     

       479
        
       

     

       480
        
       		return tx.Commit()

     

       481
        
       	case models.CommitOperationDelete:

     

       482
       -
       		log.Println("deleting reaction from pds request")

     

       483
        
       		err = db.DeleteReactionByRkey(i.Db, did, e.Commit.RKey)

     

       484
        
       	}

     

       485
        
       	if err != nil {

     
···

       546
        
       			return fmt.Errorf("invalid resource: %w", err)

     

       547
        
       		}

     

       548
        
       

     

       549
       -
       		log.Println("upserting resource from pds request")

     

       550
        
       		err = db.UpsertResource(i.Db, resource, resource.Rkey)

     

       551
        
       		if err != nil {

     

       552
        
       			tx.Rollback()

     
···

       554
        
       		}

     

       555
        
       		return tx.Commit()

     

       556
        
       	case models.CommitOperationDelete:

     

       557
       -
       		log.Println("deleting resource from pds request")

     

       558
        
       		err = db.DeleteResourceByRkey(i.Db, did, e.Commit.RKey)

     

       559
        
       	}

     

       560
        
       	if err != nil {

     
···

       626
        
       			CreatedAt:        createdAt,

     

       627
        
       		}

     

       628
        
       

     

       629
       -
       		log.Println("upserting comment from pds request")

     

       630
        
       		err = db.UpsertComment(i.Db, comment)

     

       631
        
       		if err != nil {

     

       632
        
       			tx.Rollback()

     
···

       637
        
       		if subjectDid.String() != did {

     

       638
        
       			err = db.CreateNotification(tx, subjectDid.String(), did, subjectUri.String(), db.NotificationTypeComment)

     

       639
        
       			if err != nil {

     

       640
       -
       				log.Println("failed to create notification record:", err)

     

       641
        
       			}

     

       642
        
       		}

     

       643
        
       

     
···

       645
        
       		if comment.ParentCommentUri != nil && comment.ParentCommentUri.Authority().String() != did {

     

       646
        
       			err = db.CreateNotification(tx, comment.ParentCommentUri.Authority().String(), did, parentCommentUri.String(), db.NotificationTypeReply)

     

       647
        
       			if err != nil {

     

       648
       -
       				log.Println("failed to create notification record:", err)

     

       649
        
       			}

     

       650
        
       		}

     

       651
        
       

     

       652
        
       		return tx.Commit()

     

       653
        
       	case models.CommitOperationDelete:

     

       654
       -
       		log.Println("deleting comment from pds request")

     

       655
        
       		err = db.DeleteCommentByRkey(i.Db, did, e.Commit.RKey)

     

       656
        
       	}

     

       657
        
       	if err != nil {

···

       4
        
       	"context"

     

       5
        
       	"encoding/json"

     

       6
        
       	"fmt"

     

       7
       +
       	"log/slog"

     

       8
        
       	"strings"

     

       9
        
       	"time"

     

       10
        
       

     
···

       20
        
       type Ingester struct {

     

       21
        
       	Db     db.DbWrapper

     

       22
        
       	Config *config.Config

     

       23
       +
       	Logger *slog.Logger

     

       24
        
       }

     

       25
        
       

     

       26
        
       type processFunc func(ctx context.Context, e *models.Event) error

     
···

       36
        
       			}

     

       37
        
       		}()

     

       38
        
       

     

       39
       +
       		l := i.Logger.With("kind", e.Kind)

     

       40
        
       		switch e.Kind {

     

       41
        
       		case models.EventKindCommit:

     

       42
        
       			switch e.Commit.Collection {

     

       43
        
       			case yoten.ActorProfileNSID:

     

       44
       +
       				l = l.With("handler", "ingestProfile")

     

       45
        
       				err = i.ingestProfile(e)

     

       46
        
       			case yoten.FeedSessionNSID:

     

       47
       +
       				l = l.With("handler", "ingestStudySession")

     

       48
        
       				err = i.ingestStudySession(e)

     

       49
        
       			case yoten.ActivityDefNSID:

     

       50
       +
       				l = l.With("handler", "ingestActivityDef")

     

       51
        
       				err = i.ingestActivityDef(e)

     

       52
        
       			case yoten.FeedResourceNSID:

     

       53
       +
       				l = l.With("handler", "ingestResource")

     

       54
        
       				err = i.ingestResource(e)

     

       55
        
       			case yoten.GraphFollowNSID:

     

       56
       +
       				l = l.With("handler", "ingestFollow")

     

       57
        
       				err = i.ingestFollow(e)

     

       58
        
       			case yoten.FeedReactionNSID:

     

       59
       +
       				l = l.With("handler", "ingestReaction")

     

       60
        
       				err = i.ingestReaction(e)

     

       61
        
       			case yoten.FeedCommentNSID:

     

       62
       +
       				l = l.With("handler", "ingestComment")

     

       63
        
       				err = i.ingestComment(e)

     

       64
        
       			}

     

       65
       +
       			l = i.Logger.With("nsid", e.Commit.Collection)

     

       66
        
       		}

     

       67
        
       		if err != nil {

     

       68
       +
       			l.Error("failed to ingest event", "err", err)

     

       69
        
       		}

     

       70
        
       

     

       71
        
       		return nil

     
···

       139
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       140
        
       		}

     

       141
        
       

     

       142
       +
       		i.Logger.Debug("upserting profile from pds request")

     

       143
        
       		err = db.UpsertProfile(tx, &profile)

     

       144
        
       		if err != nil {

     

       145
        
       			tx.Rollback()

     
···

       170
        
       

     

       171
        
       		date, err := time.Parse(time.RFC3339, record.Date)

     

       172
        
       		if err != nil {

     

       173
       +
       			i.Logger.Error("invalid record", "err", err)

     

       174
        
       			return err

     

       175
        
       		}

     

       176
        
       

     
···

       237
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       238
        
       		}

     

       239
        
       

     

       240
       +
       		i.Logger.Debug("upserting study session from pds request")

     

       241
        
       		err = db.UpsertStudySession(tx, &studySession, e.Commit.RKey)

     

       242
        
       		if err != nil {

     

       243
        
       			tx.Rollback()

     
···

       262
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       263
        
       		}

     

       264
        
       

     

       265
       +
       		i.Logger.Debug("deleting study session from pds request")

     

       266
        
       		err = db.DeleteStudySessionByRkey(tx, did, e.Commit.RKey)

     

       267
        
       		if err != nil {

     

       268
        
       			tx.Rollback()

     
···

       354
        
       			return fmt.Errorf("failed to start transaction: %w", err)

     

       355
        
       		}

     

       356
        
       

     

       357
       +
       		i.Logger.Debug("upserting activity def from pds request")

     

       358
        
       		err = db.UpsertActivityDef(tx, &activityDef, e.Commit.RKey)

     

       359
        
       		if err != nil {

     

       360
        
       			tx.Rollback()

     
···

       362
        
       		}

     

       363
        
       		return tx.Commit()

     

       364
        
       	case models.CommitOperationDelete:

     

       365
       +
       		i.Logger.Debug("deleting activity def from pds request")

     

       366
        
       		err = db.DeleteActivityDefByRkey(i.Db, did, e.Commit.RKey)

     

       367
        
       	}

     

       368
        
       	if err != nil {

     
···

       397
        
       

     

       398
        
       		subjectDid := record.Subject

     

       399
        
       

     

       400
       +
       		i.Logger.Debug("upserting follow from pds request")

     

       401
        
       		err = db.AddFollow(tx, did, subjectDid, e.Commit.RKey)

     

       402
        
       		if err != nil {

     

       403
        
       			tx.Rollback()

     
···

       407
        
       		subjectUri := fmt.Sprintf("at://%s/%s/%s", did, yoten.GraphFollowNSID, e.Commit.RKey)

     

       408
        
       		err = db.CreateNotification(tx, subjectDid, did, subjectUri, db.NotificationTypeFollow)

     

       409
        
       		if err != nil {

     

       410
       +
       			i.Logger.Error("failed to create notification record", "err", err)

     

       411
        
       		}

     

       412
        
       

     

       413
        
       		return tx.Commit()

     

       414
        
       	case models.CommitOperationDelete:

     

       415
       +
       		i.Logger.Debug("deleting follow from pds request")

     

       416
        
       		err = db.DeleteFollowByRkey(i.Db, did, e.Commit.RKey)

     

       417
        
       	}

     

       418
        
       	if err != nil {

     
···

       475
        
       			CreatedAt:   createdAt,

     

       476
        
       		}

     

       477
        
       

     

       478
       +
       		i.Logger.Debug("upserting reaction from pds request")

     

       479
        
       		err = db.UpsertReaction(i.Db, reactionEvent)

     

       480
        
       		if err != nil {

     

       481
        
       			tx.Rollback()

     
···

       484
        
       

     

       485
        
       		err = db.CreateNotification(tx, subjectDid.String(), did, subject.String(), db.NotificationTypeReaction)

     

       486
        
       		if err != nil {

     

       487
       +
       			i.Logger.Error("failed to create notification record", "err", err)

     

       488
        
       		}

     

       489
        
       

     

       490
        
       		return tx.Commit()

     

       491
        
       	case models.CommitOperationDelete:

     

       492
       +
       		i.Logger.Debug("deleting reaction from pds request")

     

       493
        
       		err = db.DeleteReactionByRkey(i.Db, did, e.Commit.RKey)

     

       494
        
       	}

     

       495
        
       	if err != nil {

     
···

       556
        
       			return fmt.Errorf("invalid resource: %w", err)

     

       557
        
       		}

     

       558
        
       

     

       559
       +
       		i.Logger.Debug("upserting resource from pds request")

     

       560
        
       		err = db.UpsertResource(i.Db, resource, resource.Rkey)

     

       561
        
       		if err != nil {

     

       562
        
       			tx.Rollback()

     
···

       564
        
       		}

     

       565
        
       		return tx.Commit()

     

       566
        
       	case models.CommitOperationDelete:

     

       567
       +
       		i.Logger.Debug("deleting resource from pds request")

     

       568
        
       		err = db.DeleteResourceByRkey(i.Db, did, e.Commit.RKey)

     

       569
        
       	}

     

       570
        
       	if err != nil {

     
···

       636
        
       			CreatedAt:        createdAt,

     

       637
        
       		}

     

       638
        
       

     

       639
       +
       		i.Logger.Debug("upserting comment from pds request")

     

       640
        
       		err = db.UpsertComment(i.Db, comment)

     

       641
        
       		if err != nil {

     

       642
        
       			tx.Rollback()

     
···

       647
        
       		if subjectDid.String() != did {

     

       648
        
       			err = db.CreateNotification(tx, subjectDid.String(), did, subjectUri.String(), db.NotificationTypeComment)

     

       649
        
       			if err != nil {

     

       650
       +
       				i.Logger.Error("failed to create notification record", "err", err)

     

       651
        
       			}

     

       652
        
       		}

     

       653
        
       

     
···

       655
        
       		if comment.ParentCommentUri != nil && comment.ParentCommentUri.Authority().String() != did {

     

       656
        
       			err = db.CreateNotification(tx, comment.ParentCommentUri.Authority().String(), did, parentCommentUri.String(), db.NotificationTypeReply)

     

       657
        
       			if err != nil {

     

       658
       +
       				i.Logger.Error("failed to create notification record", "err", err)

     

       659
        
       			}

     

       660
        
       		}

     

       661
        
       

     

       662
        
       		return tx.Commit()

     

       663
        
       	case models.CommitOperationDelete:

     

       664
       +
       		i.Logger.Debug("deleting comment from pds request")

     

       665
        
       		err = db.DeleteCommentByRkey(i.Db, did, e.Commit.RKey)

     

       666
        
       	}

     

       667
        
       	if err != nil {

+4 -2

internal/server/app.go

···

       85
        
       			yoten.GraphFollowNSID,

     

       86
        
       		},

     

       87
        
       		nil,

     

       88
       -
       		slog.Default(),

     

       89
        
       		wrapper,

     

       90
        
       		false,

     

       91
        
       	)

     
···

       95
        
       

     

       96
        
       	ingester := consumer.Ingester{

     

       97
        
       		Db:     wrapper,

     

       98
       -
       		Config: config}

     

       0
        
       
     

       0
        
       
     

       99
        
       	err = jc.StartJetstream(ctx, ingester.Ingest())

     

       100
        
       	if err != nil {

     

       101
        
       		return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)

···

       85
        
       			yoten.GraphFollowNSID,

     

       86
        
       		},

     

       87
        
       		nil,

     

       88
       +
       		log.SubLogger(logger, "jetstream"),

     

       89
        
       		wrapper,

     

       90
        
       		false,

     

       91
        
       	)

     
···

       95
        
       

     

       96
        
       	ingester := consumer.Ingester{

     

       97
        
       		Db:     wrapper,

     

       98
       +
       		Config: config,

     

       99
       +
       		Logger: log.SubLogger(logger, "ingester"),

     

       100
       +
       	}

     

       101
        
       	err = jc.StartJetstream(ctx, ingester.Ingest())

     

       102
        
       	if err != nil {

     

       103
        
       		return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)