+30 -20

internal/consumer/ingester.go

··· 4 4 "context" 5 5 "encoding/json" 6 6 "fmt" 7 7 - "log" 7 7 + "log/slog" 8 8 "strings" 9 9 "time" 10 10 ··· 20 20 type Ingester struct { 21 21 Db db.DbWrapper 22 22 Config *config.Config 23 23 + Logger *slog.Logger 23 24 } 24 25 25 26 type processFunc func(ctx context.Context, e *models.Event) error ··· 35 36 } 36 37 }() 37 38 39 39 + l := i.Logger.With("kind", e.Kind) 38 40 switch e.Kind { 39 41 case models.EventKindCommit: 40 42 switch e.Commit.Collection { 41 43 case yoten.ActorProfileNSID: 44 44 + l = l.With("handler", "ingestProfile") 42 45 err = i.ingestProfile(e) 43 46 case yoten.FeedSessionNSID: 47 47 + l = l.With("handler", "ingestStudySession") 44 48 err = i.ingestStudySession(e) 45 49 case yoten.ActivityDefNSID: 50 50 + l = l.With("handler", "ingestActivityDef") 46 51 err = i.ingestActivityDef(e) 47 52 case yoten.FeedResourceNSID: 53 53 + l = l.With("handler", "ingestResource") 48 54 err = i.ingestResource(e) 49 55 case yoten.GraphFollowNSID: 56 56 + l = l.With("handler", "ingestFollow") 50 57 err = i.ingestFollow(e) 51 58 case yoten.FeedReactionNSID: 59 59 + l = l.With("handler", "ingestReaction") 52 60 err = i.ingestReaction(e) 53 61 case yoten.FeedCommentNSID: 62 62 + l = l.With("handler", "ingestComment") 54 63 err = i.ingestComment(e) 55 64 } 65 65 + l = i.Logger.With("nsid", e.Commit.Collection) 56 66 } 57 67 if err != nil { 58 58 - log.Printf("failed to ingest event for collection %s: %v", e.Commit.Collection, err) 68 68 + l.Error("failed to ingest event", "err", err) 59 69 } 60 70 61 71 return nil ··· 129 139 return fmt.Errorf("failed to start transaction: %w", err) 130 140 } 131 141 132 132 - log.Printf("upserting profile '%s' from pds request", profile.Did) 142 142 + i.Logger.Debug("upserting profile from pds request") 133 143 err = db.UpsertProfile(tx, &profile) 134 144 if err != nil { 135 145 tx.Rollback() ··· 160 170 161 171 date, err := time.Parse(time.RFC3339, record.Date) 162 172 if err != nil { 163 163 - log.Printf("invalid record: %s", err) 173 173 + i.Logger.Error("invalid record", "err", err) 164 174 return err 165 175 } 166 176 ··· 227 237 return fmt.Errorf("failed to start transaction: %w", err) 228 238 } 229 239 230 230 - log.Println("upserting study session from pds request") 240 240 + i.Logger.Debug("upserting study session from pds request") 231 241 err = db.UpsertStudySession(tx, &studySession, e.Commit.RKey) 232 242 if err != nil { 233 243 tx.Rollback() ··· 252 262 return fmt.Errorf("failed to start transaction: %w", err) 253 263 } 254 264 255 255 - log.Println("deleting study session from pds request") 265 265 + i.Logger.Debug("deleting study session from pds request") 256 266 err = db.DeleteStudySessionByRkey(tx, did, e.Commit.RKey) 257 267 if err != nil { 258 268 tx.Rollback() ··· 344 354 return fmt.Errorf("failed to start transaction: %w", err) 345 355 } 346 356 347 347 - log.Println("upserting activity def from pds request") 357 357 + i.Logger.Debug("upserting activity def from pds request") 348 358 err = db.UpsertActivityDef(tx, &activityDef, e.Commit.RKey) 349 359 if err != nil { 350 360 tx.Rollback() ··· 352 362 } 353 363 return tx.Commit() 354 364 case models.CommitOperationDelete: 355 355 - log.Println("deleting activity def from pds request") 365 365 + i.Logger.Debug("deleting activity def from pds request") 356 366 err = db.DeleteActivityDefByRkey(i.Db, did, e.Commit.RKey) 357 367 } 358 368 if err != nil { ··· 387 397 388 398 subjectDid := record.Subject 389 399 390 390 - log.Println("upserting follow from pds request") 400 400 + i.Logger.Debug("upserting follow from pds request") 391 401 err = db.AddFollow(tx, did, subjectDid, e.Commit.RKey) 392 402 if err != nil { 393 403 tx.Rollback() ··· 397 407 subjectUri := fmt.Sprintf("at://%s/%s/%s", did, yoten.GraphFollowNSID, e.Commit.RKey) 398 408 err = db.CreateNotification(tx, subjectDid, did, subjectUri, db.NotificationTypeFollow) 399 409 if err != nil { 400 400 - log.Println("failed to create notification record:", err) 410 410 + i.Logger.Error("failed to create notification record", "err", err) 401 411 } 402 412 403 413 return tx.Commit() 404 414 case models.CommitOperationDelete: 405 405 - log.Println("deleting follow from pds request") 415 415 + i.Logger.Debug("deleting follow from pds request") 406 416 err = db.DeleteFollowByRkey(i.Db, did, e.Commit.RKey) 407 417 } 408 418 if err != nil { ··· 465 475 CreatedAt: createdAt, 466 476 } 467 477 468 468 - log.Println("upserting reaction from pds request") 478 478 + i.Logger.Debug("upserting reaction from pds request") 469 479 err = db.UpsertReaction(i.Db, reactionEvent) 470 480 if err != nil { 471 481 tx.Rollback() ··· 474 484 475 485 err = db.CreateNotification(tx, subjectDid.String(), did, subject.String(), db.NotificationTypeReaction) 476 486 if err != nil { 477 477 - log.Println("failed to create notification record:", err) 487 487 + i.Logger.Error("failed to create notification record", "err", err) 478 488 } 479 489 480 490 return tx.Commit() 481 491 case models.CommitOperationDelete: 482 482 - log.Println("deleting reaction from pds request") 492 492 + i.Logger.Debug("deleting reaction from pds request") 483 493 err = db.DeleteReactionByRkey(i.Db, did, e.Commit.RKey) 484 494 } 485 495 if err != nil { ··· 546 556 return fmt.Errorf("invalid resource: %w", err) 547 557 } 548 558 549 549 - log.Println("upserting resource from pds request") 559 559 + i.Logger.Debug("upserting resource from pds request") 550 560 err = db.UpsertResource(i.Db, resource, resource.Rkey) 551 561 if err != nil { 552 562 tx.Rollback() ··· 554 564 } 555 565 return tx.Commit() 556 566 case models.CommitOperationDelete: 557 557 - log.Println("deleting resource from pds request") 567 567 + i.Logger.Debug("deleting resource from pds request") 558 568 err = db.DeleteResourceByRkey(i.Db, did, e.Commit.RKey) 559 569 } 560 570 if err != nil { ··· 626 636 CreatedAt: createdAt, 627 637 } 628 638 629 629 - log.Println("upserting comment from pds request") 639 639 + i.Logger.Debug("upserting comment from pds request") 630 640 err = db.UpsertComment(i.Db, comment) 631 641 if err != nil { 632 642 tx.Rollback() ··· 637 647 if subjectDid.String() != did { 638 648 err = db.CreateNotification(tx, subjectDid.String(), did, subjectUri.String(), db.NotificationTypeComment) 639 649 if err != nil { 640 640 - log.Println("failed to create notification record:", err) 650 650 + i.Logger.Error("failed to create notification record", "err", err) 641 651 } 642 652 } 643 653 ··· 645 655 if comment.ParentCommentUri != nil && comment.ParentCommentUri.Authority().String() != did { 646 656 err = db.CreateNotification(tx, comment.ParentCommentUri.Authority().String(), did, parentCommentUri.String(), db.NotificationTypeReply) 647 657 if err != nil { 648 648 - log.Println("failed to create notification record:", err) 658 658 + i.Logger.Error("failed to create notification record", "err", err) 649 659 } 650 660 } 651 661 652 662 return tx.Commit() 653 663 case models.CommitOperationDelete: 654 654 - log.Println("deleting comment from pds request") 664 664 + i.Logger.Debug("deleting comment from pds request") 655 665 err = db.DeleteCommentByRkey(i.Db, did, e.Commit.RKey) 656 666 } 657 667 if err != nil {

+4 -2

internal/server/app.go

··· 85 85 yoten.GraphFollowNSID, 86 86 }, 87 87 nil, 88 88 - slog.Default(), 88 88 + log.SubLogger(logger, "jetstream"), 89 89 wrapper, 90 90 false, 91 91 ) ··· 95 95 96 96 ingester := consumer.Ingester{ 97 97 Db: wrapper, 98 98 - Config: config} 98 98 + Config: config, 99 99 + Logger: log.SubLogger(logger, "ingester"), 100 100 + } 99 101 err = jc.StartJetstream(ctx, ingester.Ingest()) 100 102 if err != nil { 101 103 return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)