Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
5.4 kB · 173 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174package model
import ( "context" "errors" "fmt" "time"
glex "github.com/streamplace/glex/runtime" "gorm.io/gorm" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/moderation" "stream.place/streamplace/pkg/placestream")
type Livestream struct { URI string `json:"uri" gorm:"primaryKey;column:uri"` CID string `json:"cid" gorm:"column:cid"` CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;index:idx_repo_created,priority:2"` Livestream *[]byte `json:"livestream"` RepoDID string `json:"repoDID" gorm:"column:repo_did;index:idx_repo_created,priority:1"` Repo *Repo `json:"repo,omitempty" gorm:"foreignKey:DID;references:RepoDID"` Post *FeedPost `json:"post,omitempty" gorm:"foreignKey:CID;references:PostCID"` PostCID string `json:"postCID" gorm:"column:post_cid"` PostURI string `json:"postURI" gorm:"column:post_uri;index:idx_post_uri"`}
func (ls *Livestream) ToLivestreamView() (*placestream.Livestream_LivestreamView, error) { if ls == nil || ls.Livestream == nil { return nil, fmt.Errorf("livestream record is nil") } var rec placestream.Livestream if err := glex.DecodeCBOR(*ls.Livestream, &rec); err != nil { return nil, fmt.Errorf("error decoding livestream record: %w", err) } rec.Tags = moderation.FilterTags(rec.Tags) postView := placestream.Livestream_LivestreamView{ LexiconTypeID: "place.stream.livestream#livestreamView", Cid: ls.CID, Uri: ls.URI, Author: appbsky.ActorDefs_ProfileViewBasic{ Did: ls.RepoDID, Handle: ls.Repo.Handle, }, Record: &glex.LexiconTypeDecoder{Val: &rec}, IndexedAt: time.Now().Format(time.RFC3339), } return &postView, nil}
// CreateLivestream upserts a livestream record. It used to overwrite the row on// every conflict, which meant a redelivered record -- the same bytes, the same// CID -- rewrote the row and re-announced the stream on the bus. Now an// unchanged record is a no-op and only a genuinely new version is stored.func (m *DBModel) CreateLivestream(ctx context.Context, ls *Livestream) error { return createOrVerify(ctx, m, ls, map[string]any{"uri": ls.URI})}
func (m *DBModel) GetLivestream(uri string) (*Livestream, error) { var livestream Livestream err := m.DB. Preload("Repo"). Preload("Post"). Where("uri = ?", uri). First(&livestream).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil } if err != nil { return nil, fmt.Errorf("error retrieving livestream by uri: %w", err) } return &livestream, nil}
// GetLatestLivestreamForRepo returns the most recent livestream for a given repo DIDfunc (m *DBModel) GetLatestLivestreamForRepo(repoDID string) (*Livestream, error) { var livestream Livestream err := m.DB. Preload("Repo"). Preload("Post"). Where("repo_did = ?", repoDID). Order("created_at DESC"). First(&livestream).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil } if err != nil { return nil, fmt.Errorf("error retrieving latest livestream: %w", err) } return &livestream, nil}
func (m *DBModel) GetLivestreamByPostURI(postURI string) (*Livestream, error) { var livestream Livestream err := m.DB. Preload("Repo"). Preload("Post"). Where("post_uri = ?", postURI). First(&livestream).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil } if err != nil { return nil, fmt.Errorf("error retrieving livestream by postURI: %w", err) } return &livestream, nil}
// Get the latest livestreams for a given list of repo DIDsfunc (m *DBModel) GetLatestLivestreams(limit int, before *time.Time, dids []string) ([]Livestream, error) { var recentLivestreams []Livestream now := time.Now().UTC()
if len(dids) == 0 { return []Livestream{}, nil }
// Subquery to get the most recent livestream for each repo_did subQuery := m.DB. Table("livestreams"). Select("MAX(created_at) as max_created_at, repo_did"). Where("repo_did IN ?", dids). Group("repo_did")
mainQuery := m.DB. Table("livestreams"). Select("livestreams.*"). Joins("JOIN (?) as sq ON livestreams.repo_did = sq.repo_did AND livestreams.created_at = sq.max_created_at", subQuery). Where("livestreams.repo_did IN ?", dids). // exclude livestreams with !hide label on the record Where("NOT EXISTS (?)", m.DB.Table("labels"). Select("1"). Where("labels.uri = livestreams.uri"). Where("labels.val = ?", "!hide"). Where("labels.neg = ?", false). Where("(labels.exp IS NULL OR labels.exp > ?)", now), ). // exclude livestreams with !hide label on the user Where("NOT EXISTS (?)", m.DB.Table("labels"). Select("1"). Where("labels.uri = livestreams.repo_did"). Where("labels.val = ?", "!hide"). Where("labels.neg = ?", false). Where("(labels.exp IS NULL OR labels.exp > ?)", now), )
if before != nil { mainQuery = mainQuery.Where("livestreams.created_at < ?", *before) }
mainQuery = mainQuery. Order("livestreams.created_at DESC"). Limit(limit). Preload("Repo")
err := mainQuery.Find(&recentLivestreams).Error
if errors.Is(err, gorm.ErrRecordNotFound) { return nil, nil }
if err != nil { return nil, fmt.Errorf("error fetching recent livestreams: %w", err) }
if len(recentLivestreams) == 0 { return nil, nil }
return recentLivestreams, nil}