diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 9dfffe99..5427219c 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -428,7 +428,7 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { return err } } - swarm, err := iroh_replicator.NewSwarm(ctx, cli.Tickets, secret, topic, mm, b) + swarm, err := iroh_replicator.NewSwarm(ctx, cli.Tickets, secret, topic, mm, b, mod) if err != nil { return err } diff --git a/pkg/config/config.go b/pkg/config/config.go index d4c3f1dc..8a917f79 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -279,7 +279,7 @@ var GormLogger = slogGorm.New( slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{ TimeFormat: time.RFC3339, })), - // slogGorm.WithTraceAll(), + slogGorm.WithTraceAll(), ) func (cli *CLI) Parse(fs *flag.FlagSet, args []string) error { diff --git a/pkg/model/broadcast_origin.go b/pkg/model/broadcast_origin.go index 66f5e5b1..5bc58890 100644 --- a/pkg/model/broadcast_origin.go +++ b/pkg/model/broadcast_origin.go @@ -6,7 +6,10 @@ import ( "fmt" "time" + "github.com/bluesky-social/indigo/api/bsky" "github.com/bluesky-social/indigo/atproto/syntax" + lexutil "github.com/bluesky-social/indigo/lex/util" + "stream.place/streamplace/pkg/aqtime" "stream.place/streamplace/pkg/spid" "stream.place/streamplace/pkg/streamplace" ) @@ -16,10 +19,10 @@ type BroadcastOrigin struct { CID string `gorm:"column:cid"` RepoDID string `gorm:"column:repo_did"` Repo *Repo `gorm:"foreignKey:DID;references:RepoDID"` - StreamerRepoDID string `gorm:"column:streamer_repo_did;index:idx_streamer_repo_did_updated_at,priority:1"` + StreamerRepoDID string `gorm:"column:streamer_repo_did;index:idx_streamer_repo_did_indexed_at,priority:1"` StreamerRepo *Repo `gorm:"foreignKey:DID;references:StreamerRepoDID"` - ServerRepoDID string `gorm:"column:server_repo_did;index:idx_server_repo_did_updated_at,priority:1"` - UpdatedAt time.Time `gorm:"column:updated_at;index:idx_streamer_repo_did_updated_at,priority:2;index:idx_server_repo_did_updated_at,priority:2"` + ServerRepoDID string `gorm:"column:server_repo_did;index:idx_server_repo_did_indexed_at,priority:1"` + IndexedAt time.Time `gorm:"column:indexed_at;index:idx_streamer_repo_did_indexed_at,priority:2;index:idx_server_repo_did_indexed_at,priority:2"` Record []byte `gorm:"column:record"` } @@ -27,6 +30,21 @@ func (bo *BroadcastOrigin) TableName() string { return "broadcast_origins" } +func (bo *BroadcastOrigin) ToBroadcastOriginView() (*streamplace.BroadcastDefs_BroadcastOriginView, error) { + rec, err := lexutil.CborDecodeValue(bo.Record) + if err != nil { + return nil, fmt.Errorf("error decoding broadcast origin: %w", err) + } + return &streamplace.BroadcastDefs_BroadcastOriginView{ + Author: &bsky.ActorDefs_ProfileViewBasic{ + Did: bo.StreamerRepoDID, + }, + Cid: bo.CID, + Record: &lexutil.LexiconTypeDecoder{Val: rec}, + Uri: bo.URI, + }, nil +} + func (m *DBModel) UpdateBroadcastOrigin(ctx context.Context, origin *streamplace.BroadcastOrigin, aturi syntax.ATURI) error { repoDID := aturi.Authority().String() cid, err := spid.GetCID(origin) @@ -42,13 +60,37 @@ func (m *DBModel) UpdateBroadcastOrigin(ctx context.Context, origin *streamplace if err != nil { return fmt.Errorf("failed to marshal origin: %w", err) } + aqt := aqtime.FromTime(time.Now().UTC()) + bo := &BroadcastOrigin{ URI: validATURI, CID: cid.String(), StreamerRepoDID: origin.Streamer, ServerRepoDID: origin.Server, - UpdatedAt: time.Now(), + IndexedAt: aqt.Time().UTC(), Record: buf.Bytes(), } return m.DB.Save(bo).Error } + +func (m *DBModel) GetRecentBroadcastOrigins(ctx context.Context) ([]*streamplace.BroadcastDefs_BroadcastOriginView, error) { + now := time.Now() + oneMinuteAgo := now.Add(-1 * time.Minute) + + var origins []*BroadcastOrigin + err := m.DB. + Where("indexed_at >= ?", oneMinuteAgo.UTC()). + Find(&origins).Error + if err != nil { + return nil, err + } + views := make([]*streamplace.BroadcastDefs_BroadcastOriginView, len(origins)) + for i, o := range origins { + view, err := o.ToBroadcastOriginView() + if err != nil { + return nil, err + } + views[i] = view + } + return views, nil +} diff --git a/pkg/model/model.go b/pkg/model/model.go index f18b94af..91056c38 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -99,6 +99,7 @@ type Model interface { GetActiveLabels(uri string) ([]*comatproto.LabelDefs_Label, error) UpdateBroadcastOrigin(ctx context.Context, origin *streamplace.BroadcastOrigin, aturi syntax.ATURI) error + GetRecentBroadcastOrigins(ctx context.Context) ([]*streamplace.BroadcastDefs_BroadcastOriginView, error) } var DBRevision = 2 diff --git a/pkg/replication/iroh_replicator/kv.go b/pkg/replication/iroh_replicator/kv.go index 0d3c6854..41604eac 100644 --- a/pkg/replication/iroh_replicator/kv.go +++ b/pkg/replication/iroh_replicator/kv.go @@ -15,6 +15,7 @@ import ( "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" + "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/streamplace" ) @@ -30,6 +31,7 @@ type IrohSwarm struct { handleDataScoped func(topic string, data []byte) bus *bus.Bus originMutex sync.Mutex + mod model.Model } // A message saying "hey I ingested node data at this time" @@ -38,7 +40,7 @@ type OriginInfo struct { Time string `json:"time"` } -func NewSwarm(ctx context.Context, tickets []string, secret []byte, topic []byte, mm *media.MediaManager, bus *bus.Bus) (*IrohSwarm, error) { +func NewSwarm(ctx context.Context, tickets []string, secret []byte, topic []byte, mm *media.MediaManager, bus *bus.Bus, mod model.Model) (*IrohSwarm, error) { ctx = log.WithLogValues(ctx, "func", "StartKV") if topic == nil { @@ -61,6 +63,7 @@ func NewSwarm(ctx context.Context, tickets []string, secret []byte, topic []byte mm: mm, activeSubs: make(map[string]*OriginInfo), bus: bus, + mod: mod, } // workaround to get context into the HandleData callback @@ -183,47 +186,63 @@ func (swarm *IrohSwarm) startKV(ctx context.Context) error { // subscribe to all streams func (swarm *IrohSwarm) startBusSubscribe(ctx context.Context) error { + // start subscription first so we're buffering new origins + busCh := swarm.bus.Subscribe("") + originViews, err := swarm.mod.GetRecentBroadcastOrigins(ctx) + if err != nil { + return fmt.Errorf("failed to get recent broadcast origins: %w", err) + } + for _, view := range originViews { + err = swarm.handleOriginMessage(ctx, view) + if err != nil { + log.Error(ctx, "could not check origin", "error", err) + } + } + log.Log(ctx, "Resumed recent broadcast origins", "count", len(originViews)) for { select { case <-ctx.Done(): return ctx.Err() - case msg := <-swarm.bus.Subscribe(""): + case msg := <-busCh: if view, ok := msg.(*streamplace.BroadcastDefs_BroadcastOriginView); ok { log.Debug(ctx, "got broadcast origin view", "view", view) - origin, ok := view.Record.Val.(*streamplace.BroadcastOrigin) - if !ok { - log.Error(ctx, "record is not a BroadcastOrigin", "record", view.Record) - continue - } - if view.Author.Did != origin.Streamer { - // currently, only streamers are allowed to advertise origins - continue - } - if origin.IrohTicket == nil { - log.Error(ctx, "origin has no iroh ticket", "origin", origin) - continue - } - pubKey, err := iroh_streamplace.NodeIdFromTicket(*origin.IrohTicket) - if err != nil { - log.Error(ctx, "could not get node id from ticket", "error", err) - continue - } - err = swarm.Node.AddTickets([]string{*origin.IrohTicket}) + err = swarm.handleOriginMessage(ctx, view) if err != nil { - log.Error(ctx, "could not add tickets", "error", err) - continue - } - pubKeyStr := pubKey.String() - err = swarm.checkOrigins(ctx, origin.Streamer, pubKeyStr) - if err != nil { - log.Error(ctx, "could not check origin", "error", err) - continue + log.Error(ctx, "could not handle origin message", "error", err) } } } } } +func (swarm *IrohSwarm) handleOriginMessage(ctx context.Context, view *streamplace.BroadcastDefs_BroadcastOriginView) error { + origin, ok := view.Record.Val.(*streamplace.BroadcastOrigin) + if !ok { + return fmt.Errorf("record is not a BroadcastOrigin") + } + if view.Author.Did != origin.Streamer { + // currently, only streamers are allowed to advertise origins + return nil + } + if origin.IrohTicket == nil { + return fmt.Errorf("origin has no iroh ticket") + } + pubKey, err := iroh_streamplace.NodeIdFromTicket(*origin.IrohTicket) + if err != nil { + return fmt.Errorf("could not get node id from ticket: %w", err) + } + err = swarm.Node.AddTickets([]string{*origin.IrohTicket}) + if err != nil { + return fmt.Errorf("could not add tickets: %w", err) + } + pubKeyStr := pubKey.String() + err = swarm.checkOrigins(ctx, origin.Streamer, pubKeyStr) + if err != nil { + return fmt.Errorf("could not check origin: %w", err) + } + return nil +} + func (swarm *IrohSwarm) checkOrigins(ctx context.Context, streamer string, nodeID string) error { swarm.originMutex.Lock() defer swarm.originMutex.Unlock()