Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
7.9 kB · 216 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217package statedb
import ( "bytes" "fmt" "time"
"github.com/bluesky-social/indigo/util" glex "github.com/streamplace/glex/runtime" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/spid")
const MAX_MULTISTREAM_TARGETS = 100const MAX_ACTIVE_MULTISTREAM_TARGETS = 5
type MultistreamTarget struct { URI string `gorm:"column:uri;primarykey"` CID string `gorm:"column:cid;not null"` Active bool `gorm:"column:active"` RepoDID string `gorm:"column:repo_did;not null;index"` MultistreamTarget []byte `gorm:"column:record"`}
func (m *MultistreamTarget) TableName() string { return "multistream_targets"}
func (state *StatefulDB) CreateMultistreamTarget(input placestream.MultistreamCreateTarget_Input, repoDID string) (placestream.MultistreamDefs_TargetView, error) { // Check total targets limit var totalCount int64 err := state.DB.Model(&MultistreamTarget{}).Where("repo_did = ?", repoDID).Count(&totalCount).Error if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to count existing targets: %w", err) } if totalCount >= MAX_MULTISTREAM_TARGETS { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("maximum number of multistream targets (%d) reached", MAX_MULTISTREAM_TARGETS) }
// Check active targets limit if this target is active if input.MultistreamTarget.Active { var activeCount int64 err := state.DB.Model(&MultistreamTarget{}).Where("repo_did = ? AND active = ?", repoDID, true).Count(&activeCount).Error if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to count active targets: %w", err) } if activeCount >= MAX_ACTIVE_MULTISTREAM_TARGETS { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("maximum number of active multistream targets (%d) reached", MAX_ACTIVE_MULTISTREAM_TARGETS) } }
// this URI is, of course, a LIE tid := spid.TIDClock.Next() uri := fmt.Sprintf("at://%s/place.stream.multistream.target/%s", repoDID, tid.String())
cid, err := spid.GetCID(&input.MultistreamTarget) if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to get CID: %w", err) }
buf := bytes.Buffer{} err = input.MultistreamTarget.MarshalCBOR(&buf) if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to marshal multistream target: %w", err) }
dbTarget := &MultistreamTarget{ URI: uri, CID: cid.String(), RepoDID: repoDID, MultistreamTarget: buf.Bytes(), Active: input.MultistreamTarget.Active, } err = state.DB.Create(dbTarget).Error if err != nil { return placestream.MultistreamDefs_TargetView{}, err } return placestream.MultistreamDefs_TargetView{ Uri: uri, Cid: cid.String(), Record: &glex.LexiconTypeDecoder{Val: &input.MultistreamTarget}, }, nil}
func (state *StatefulDB) GetMultistreamTarget(uri string) (placestream.MultistreamDefs_TargetView, error) { return placestream.MultistreamDefs_TargetView{}, nil}
type TargetWithEvent struct { MultistreamTarget LatestEventID *string `gorm:"column:latest_event_id"` LatestEventStatus *string `gorm:"column:latest_event_status"` LatestEventMessage *string `gorm:"column:latest_event_message"` LatestEventCreatedAt *time.Time `gorm:"column:latest_event_created_at"`}
func (state *StatefulDB) ListMultistreamTargets(repoDID string, limit int, offset int, active *bool) ([]placestream.MultistreamDefs_TargetView, error) {
var targets []TargetWithEvent query := state.DB.Table("multistream_targets"). Select("multistream_targets.*, me.id as latest_event_id, me.status as latest_event_status, me.message as latest_event_message, me.created_at as latest_event_created_at"). Joins(`LEFT JOIN multistream_events me ON multistream_targets.uri = me.target_uri AND me.created_at = (SELECT MAX(created_at) FROM multistream_events WHERE target_uri = multistream_targets.uri)`). Where("repo_did = ?", repoDID)
if active != nil { query = query.Where("active = ?", *active) }
err := query.Limit(limit). Offset(offset). Order("uri ASC"). Find(&targets).Error if err != nil { return nil, fmt.Errorf("failed to list multistream targets: %w", err) }
result := make([]placestream.MultistreamDefs_TargetView, len(targets)) for i, target := range targets { var multistreamTarget placestream.MultistreamTarget err = multistreamTarget.UnmarshalCBOR(bytes.NewReader(target.MultistreamTarget.MultistreamTarget)) if err != nil { return nil, fmt.Errorf("failed to unmarshal multistream target: %w", err) } cid, err := spid.GetCID(&multistreamTarget) if err != nil { return nil, fmt.Errorf("failed to get CID: %w", err) }
targetView := placestream.MultistreamDefs_TargetView{ Uri: target.URI, Cid: cid.String(), Record: &glex.LexiconTypeDecoder{Val: &multistreamTarget}, }
// Add the latest event if it exists if target.LatestEventID != nil { event := placestream.MultistreamDefs_Event{ Status: *target.LatestEventStatus, Message: *target.LatestEventMessage, CreatedAt: target.LatestEventCreatedAt.Format(util.ISO8601), } targetView.LatestEvent = &event }
result[i] = targetView }
return result, nil}
func (state *StatefulDB) UpdateMultistreamTarget(uri string, input placestream.MultistreamPutTarget_Input) (placestream.MultistreamDefs_TargetView, error) { // Get the current target to check repo ownership and current active status var currentTarget MultistreamTarget err := state.DB.Where("uri = ?", uri).First(¤tTarget).Error if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("multistream target not found") }
// If updating to active and wasn't previously active, check active targets limit if input.MultistreamTarget.Active && !currentTarget.Active { var activeCount int64 err := state.DB.Model(&MultistreamTarget{}).Where("repo_did = ? AND active = ?", currentTarget.RepoDID, true).Count(&activeCount).Error if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to count active targets: %w", err) } if activeCount >= MAX_ACTIVE_MULTISTREAM_TARGETS { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("maximum number of active multistream targets (%d) reached", MAX_ACTIVE_MULTISTREAM_TARGETS) } }
// Get CID for the updated target cid, err := spid.GetCID(&input.MultistreamTarget) if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to get CID: %w", err) }
// Marshal the target data buf := bytes.Buffer{} err = input.MultistreamTarget.MarshalCBOR(&buf) if err != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to marshal multistream target: %w", err) }
// Update the database record updates := map[string]interface{}{ "cid": cid.String(), "record": buf.Bytes(), "active": input.MultistreamTarget.Active, }
result := state.DB.Model(&MultistreamTarget{}).Where("uri = ?", uri).Updates(updates) if result.Error != nil { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("failed to update multistream target: %w", result.Error) } if result.RowsAffected == 0 { return placestream.MultistreamDefs_TargetView{}, fmt.Errorf("multistream target not found") }
return placestream.MultistreamDefs_TargetView{ Uri: uri, Cid: cid.String(), Record: &glex.LexiconTypeDecoder{Val: &input.MultistreamTarget}, }, nil}
func (state *StatefulDB) DeleteMultistreamTarget(uri string) error { result := state.DB.Where("uri = ?", uri).Delete(&MultistreamTarget{}) if result.Error != nil { return fmt.Errorf("failed to delete multistream target: %w", result.Error) } if result.RowsAffected == 0 { return fmt.Errorf("multistream target not found") } return nil}