Something went wrong. Try again.
An implementation of the ATProto statusphere example app but in Go
Something went wrong. Try again.
statusphere-go consumer.go
2.9 kB · 123 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124package statusphere
import ( "context" "encoding/json"
"fmt" "log/slog" "time"
"github.com/bluesky-social/jetstream/pkg/client" "github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential" "github.com/bluesky-social/jetstream/pkg/models")
type consumer struct { cfg *client.ClientConfig handler handler logger *slog.Logger}
func NewConsumer(jsAddr string, logger *slog.Logger, store HandlerStore) *consumer { cfg := client.DefaultClientConfig() if jsAddr != "" { cfg.WebsocketURL = jsAddr } cfg.WantedCollections = []string{ "xyz.statusphere.status", } cfg.WantedDids = []string{}
return &consumer{ cfg: cfg, logger: logger, handler: handler{ store: store, }, }}
func (c *consumer) Consume(ctx context.Context) error { scheduler := sequential.NewScheduler("jetstream_localdev", c.logger, c.handler.HandleEvent) defer scheduler.Shutdown()
client, err := client.NewClient(c.cfg, c.logger, scheduler) if err != nil { return fmt.Errorf("failed to create client: %w", err) }
cursor, err := c.handler.store.GetCursor(ctx) // if cursor can't be fetched, just start from a couple days ago. if err != nil || cursor == 0 { cursor = time.Now().Add(-time.Hour * 48).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) }
slog.Info("stopping consume") return nil}
type HandlerStore interface { CreateStatus(status Status) error SaveCursor(ctx context.Context, cursor int64) error GetCursor(ctx context.Context) (int64, error)}
type handler struct { store HandlerStore}
func (h *handler) HandleEvent(ctx context.Context, event *models.Event) error { if event.Commit == nil { return nil }
defer func() { err := h.store.SaveCursor(ctx, event.TimeUS) if err != nil { slog.Error("failed to save cursor", "error", err) } }()
switch event.Commit.Operation { case models.CommitOperationCreate: return h.handleCreateEvent(ctx, event) default: return nil }}
type StatusRecord struct { Status string `json:"status"` CreatedAt time.Time `json:"createdAt"`}
func (h *handler) handleCreateEvent(_ context.Context, event *models.Event) error { var statusRecord StatusRecord if err := json.Unmarshal(event.Commit.Record, &statusRecord); err != nil { slog.Error("unmarshal record", "error", err) return nil }
uri := fmt.Sprintf("at://%s/%s/%s", event.Did, event.Commit.Collection, event.Commit.RKey)
status := Status{ URI: uri, Did: event.Did, Status: statusRecord.Status, CreatedAt: statusRecord.CreatedAt.UnixMilli(), IndexedAt: time.Now().UnixMilli(), } err := h.store.CreateStatus(status) if err != nil { slog.Error("failed to store status", "error", err) }
return nil}