+16 -1

consumer.go

··· 47 47 return fmt.Errorf("failed to create client: %w", err) 48 48 } 49 49 50 50 - cursor := time.Now().Add(1 * -time.Minute).UnixMicro() 50 50 + cursor, err := c.handler.store.GetCursor(ctx) 51 51 + // if cursor can't be fetched, just start from a couple days ago. 52 52 + if err != nil || cursor == 0 { 53 53 + cursor = time.Now().Add(-time.Hour * 48).UnixMicro() 54 54 + } 55 55 + 56 56 + slog.Info("starting from cursor", "time", time.UnixMicro(cursor), "cursor", cursor) 51 57 52 58 if err := client.ConnectAndRead(ctx, &cursor); err != nil { 53 59 return fmt.Errorf("connect and read: %w", err) ··· 59 65 60 66 type HandlerStore interface { 61 67 CreateStatus(status Status) error 68 68 + SaveCursor(ctx context.Context, cursor int64) error 69 69 + GetCursor(ctx context.Context) (int64, error) 62 70 } 63 71 64 72 type handler struct { ··· 69 77 if event.Commit == nil { 70 78 return nil 71 79 } 80 80 + 81 81 + defer func() { 82 82 + err := h.store.SaveCursor(ctx, event.TimeUS) 83 83 + if err != nil { 84 84 + slog.Error("failed to save cursor", "error", err) 85 85 + } 86 86 + }() 72 87 73 88 switch event.Commit.Operation { 74 89 case models.CommitOperationCreate:

+5

database/database.go

··· 52 52 return nil, fmt.Errorf("creating profile table: %w", err) 53 53 } 54 54 55 55 + err = createJetstreamTable(db) 56 56 + if err != nil { 57 57 + return nil, fmt.Errorf("creating jetstream table: %w", err) 58 58 + } 59 59 + 55 60 return &DB{db: db}, nil 56 61 } 57 62

+58

database/jetstream_cursor.go

··· 1 1 + package database 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "database/sql" 6 6 + "fmt" 7 7 + "log/slog" 8 8 + ) 9 9 + 10 10 + func createJetstreamTable(db *sql.DB) error { 11 11 + createJetstreamTableSQL := `CREATE TABLE IF NOT EXISTS jetstream ( 12 12 + "id" integer NOT NULL PRIMARY KEY, 13 13 + "cursor" INTEGER, 14 14 + UNIQUE(id) 15 15 + );` 16 16 + 17 17 + slog.Info("Create jetstream table...") 18 18 + statement, err := db.Prepare(createJetstreamTableSQL) 19 19 + if err != nil { 20 20 + return fmt.Errorf("prepare DB statement to create jetstream table: %w", err) 21 21 + } 22 22 + _, err = statement.Exec() 23 23 + if err != nil { 24 24 + return fmt.Errorf("exec sql statement to create jetstream table: %w", err) 25 25 + } 26 26 + slog.Info("jetstream table created") 27 27 + 28 28 + return nil 29 29 + } 30 30 + 31 31 + func (d *DB) SaveCursor(ctx context.Context, cursor int64) error { 32 32 + sql := `INSERT INTO jetstream (id, cursor) VALUES (1, ?) ON CONFLICT(id) DO UPDATE SET cursor = ?;` 33 33 + _, err := d.db.Exec(sql, cursor, cursor) 34 34 + if err != nil { 35 35 + return fmt.Errorf("exec insert or update cursor: %w", err) 36 36 + } 37 37 + 38 38 + return nil 39 39 + } 40 40 + 41 41 + func (d *DB) GetCursor(ctx context.Context) (int64, error) { 42 42 + sql := "SELECT cursor FROM jetstream where id = 1;" 43 43 + rows, err := d.db.Query(sql) 44 44 + if err != nil { 45 45 + return 0, fmt.Errorf("run query to get cursor: %w", err) 46 46 + } 47 47 + defer rows.Close() 48 48 + 49 49 + cursor := 0 50 50 + for rows.Next() { 51 51 + if err := rows.Scan(&cursor); err != nil { 52 52 + return 0, fmt.Errorf("scan row: %w", err) 53 53 + } 54 54 + 55 55 + return int64(cursor), nil 56 56 + } 57 57 + return 0, fmt.Errorf("not found") 58 58 + }