track the jetstream cursor

Signed-off-by: Will Andrews <did:plc:dadhhalkfcq3gucaq25hjqon>

consumer.go

···
 		logger: logger,
 		handler: handler{
 			store: store,
+			did:   did,
 		},
 	}
 }
···
 		return fmt.Errorf("failed to create client: %w", err)
 	}
 
-	cursor := time.Now().Add(1 * -time.Minute).UnixMicro()
+	cursor, err := c.handler.store.GetCursor(ctx, c.handler.did)
+	// if error or not found set to be around the time this app was create so that it starts from the begining
+	// of when the type of records were first created
+	if err != nil || cursor == 0 {
+		cursor = time.Date(2025, time.October, 5, 12, 0, 0, 0, time.UTC).UnixMicro()
+	}
+
+	slog.Info("starting from cursor", "time", time.UnixMicro(cursor), "cursor", cursor)
 
 	if err := client.ConnectAndRead(ctx, &cursor); err != nil {
 		return fmt.Errorf("connect and read: %w", err)
···
 type HandlerStore interface {
 	CreateURL(id, url, did, originHost string, createdAt int64) error
 	DeleteURL(id, did string) error
+	SaveCursor(ctx context.Context, did string, cursor int64) error
+	GetCursor(ctx context.Context, did string) (int64, error)
 }
 
 type handler struct {
 	store HandlerStore
+	did   string
 }
 
 func (h *handler) HandleEvent(ctx context.Context, event *models.Event) error {
+	if event == nil {
+		return nil
+	}
+
+	defer func() {
+		err := h.store.SaveCursor(ctx, h.did, event.TimeUS)
+		if err != nil {
+			slog.Error("failed to save cursor", "error", err)
+		}
+	}()
 	if event.Commit == nil {
 		return nil
 	}
+
+	slog.Info("handle event")
 
 	switch event.Commit.Operation {
 	case models.CommitOperationCreate:

database/database.go

···
 	if err != nil {
 		return nil, fmt.Errorf("creating status table: %w", err)
 	}
+	err = createJetstreamTable(db)
+	if err != nil {
+		return nil, fmt.Errorf("creating jetstream table: %w", err)
+	}
 
 	return &DB{db: db}, nil
 }

database/jetstream_cursor.go

···
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+)
+
+func createJetstreamTable(db *sql.DB) error {
+	createJetstreamTableSQL := `CREATE TABLE IF NOT EXISTS jetstream (
+		"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
+		"did" TEXT,
+		"cursor" INTEGER,
+		UNIQUE(did)
+	  );`
+
+	slog.Info("Create jetstream table...")
+	statement, err := db.Prepare(createJetstreamTableSQL)
+	if err != nil {
+		return fmt.Errorf("prepare DB statement to create jetstream table: %w", err)
+	}
+	_, err = statement.Exec()
+	if err != nil {
+		return fmt.Errorf("exec sql statement to create jetstream table: %w", err)
+	}
+	slog.Info("jetstream table created")
+
+	return nil
+}
+
+func (d *DB) SaveCursor(ctx context.Context, did string, cursor int64) error {
+	sql := `INSERT INTO jetstream (did, cursor) VALUES (?, ?) ON CONFLICT(did) DO UPDATE SET cursor = ?;`
+	_, err := d.db.Exec(sql, did, cursor, cursor)
+	if err != nil {
+		return fmt.Errorf("exec insert or update cursor: %w", err)
+	}
+
+	return nil
+}
+
+func (d *DB) GetCursor(ctx context.Context, did string) (int64, error) {
+	sql := "SELECT cursor FROM jetstream where did = ?;"
+	rows, err := d.db.Query(sql, did)
+	if err != nil {
+		return 0, fmt.Errorf("run query to get cursor: %w", err)
+	}
+	defer rows.Close()
+
+	cursor := 0
+	for rows.Next() {
+		if err := rows.Scan(&cursor); err != nil {
+			return 0, fmt.Errorf("scan row: %w", err)
+		}
+
+		return int64(cursor), nil
+	}
+	return 0, fmt.Errorf("not found")
+}

docker-compose.yaml

···
 services:
-  knot:
+  at-shorter:
+    platform: linux/amd64
     container_name: at-shorter
     image: willdot/at-shorter:latest
     environment:
-      ENV_LOCATION: "/data/at-shorter.env"
+      ENV_LOCATION: "/app/data/at-shorter.env"
     volumes:
       - ./data:/app/data
     ports: