model: much faster latest-segment detection

pkg/config/config.go

···
 	slogGorm.WithTraceAll(),
 )
 
+func DisableSQLLogging() {
+	GormLogger = slogGorm.New(
+		slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
+			TimeFormat: time.RFC3339,
+		})),
+	)
+}
+
+func EnableSQLLogging() {
+	GormLogger = slogGorm.New(
+		slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
+			TimeFormat: time.RFC3339,
+		})),
+		slogGorm.WithTraceAll(),
+	)
+}
+
 func (cli *CLI) Parse(fs *flag.FlagSet, args []string) error {
 	err := ff.Parse(
 		fs, args,
···
 		*dest = strings.Replace(*dest, SPDataDir, cli.DataDir, 1)
 	}
 	if !cli.SQLLogging {
-		GormLogger = slogGorm.New(
-			slogGorm.WithHandler(tint.NewHandler(os.Stderr, &tint.Options{
-				TimeFormat: time.RFC3339,
-			})),
-		)
+		DisableSQLLogging()
+	} else {
+		EnableSQLLogging()
 	}
 	if cli.XXDeprecatedPublicHost != "" && cli.BroadcasterHost == "" {
 		log.Warn(context.Background(), "public-host is deprecated, use broadcaster-host or server-host instead as appropriate")

pkg/model/segment.go

···
 	ID                 string               `json:"id"                   gorm:"primaryKey"`
 	SigningKeyDID      string               `json:"signingKeyDID"        gorm:"column:signing_key_did"`
 	SigningKey         *SigningKey          `json:"signingKey,omitempty" gorm:"foreignKey:DID;references:SigningKeyDID"`
-	StartTime          time.Time            `json:"startTime"            gorm:"index:latest_segments"`
-	RepoDID            string               `json:"repoDID"              gorm:"index:latest_segments;column:repo_did"`
+	StartTime          time.Time            `json:"startTime"            gorm:"index:latest_segments,priority:2;index:start_time"`
+	RepoDID            string               `json:"repoDID"              gorm:"index:latest_segments,priority:1;column:repo_did"`
 	Repo               *Repo                `json:"repo,omitempty"       gorm:"foreignKey:DID;references:RepoDID"`
 	Title              string               `json:"title"`
 	Size               int                  `json:"size"                gorm:"column:size"`
···
 
 	err := m.DB.Table("segments").
 		Select("segments.*").
-		Where("id IN (?)",
-			m.DB.Table("segments").
-				Select("id").
-				Where("(repo_did, start_time) IN (?)",
-					m.DB.Table("segments").
-						Select("repo_did, MAX(start_time)").
-						Group("repo_did"))).
+		Where("start_time > ?", thirtySecondsAgo.UTC()).
 		Order("start_time DESC").
-		Joins("JOIN repos ON segments.repo_did = repos.did").
-		Preload("Repo").
 		Find(&segments).Error
-
 	if err != nil {
 		return nil, err
 	}
···
 		return []Segment{}, nil
 	}
 
-	filteredSegments := []Segment{}
+	segmentMap := make(map[string]Segment)
 	for _, seg := range segments {
-		if seg.StartTime.After(thirtySecondsAgo) {
-			filteredSegments = append(filteredSegments, seg)
+		prev, ok := segmentMap[seg.RepoDID]
+		if !ok {
+			segmentMap[seg.RepoDID] = seg
+		} else {
+			if seg.StartTime.After(prev.StartTime) {
+				segmentMap[seg.RepoDID] = seg
+			}
 		}
 	}
 
+	filteredSegments := []Segment{}
+	for _, seg := range segmentMap {
+		filteredSegments = append(filteredSegments, seg)
+	}
+
 	return filteredSegments, nil
 }
 
···
 		return nil, err
 	}
 	return segs, nil
-}
-
-func (m *DBModel) GetLiveUsers() ([]Segment, error) {
-	var liveUsers []Segment
-	thirtySecondsAgo := aqtime.FromTime(time.Now().Add(-30 * time.Second)).Time()
-
-	err := m.DB.Model(&Segment{}).
-		Preload("Repo").
-		Where("start_time >= ?", thirtySecondsAgo).
-		Where("start_time = (SELECT MAX(start_time) FROM segments s2 WHERE s2.repo_did = segments.repo_did)").
-		Order("start_time DESC").
-		Find(&liveUsers).Error
-
-	if err != nil {
-		return nil, err
-	}
-	if liveUsers == nil {
-		return []Segment{}, nil
-	}
-
-	return liveUsers, nil
 }
 
 func (m *DBModel) GetSegment(id string) (*Segment, error) {

pkg/model/segment_test.go

···
 package model
 
-// func TestSegmentCleaner(t *testing.T) {
-// 	db, err := MakeDB(":memory:")
-// 	require.NoError(t, err)
-// 	// Create a model instance
-// 	model := db.(*DBModel)
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
 
-// 	// Create a repo for testing
-// 	repo := &Repo{
-// 		DID: "did:plc:test123",
-// 	}
-// 	err = model.DB.Create(repo).Error
-// 	require.NoError(t, err)
+	"github.com/stretchr/testify/require"
+	"stream.place/streamplace/pkg/config"
+)
 
-// 	// Create 100 segments with timestamps 1 hour ago, each one second apart
-// 	baseTime := time.Now().Add(-1 * time.Hour)
-// 	for i := 0; i < 100; i++ {
-// 		segment := &Segment{
-// 			ID:        fmt.Sprintf("segment-%d", i),
-// 			RepoDID:   repo.DID,
-// 			StartTime: baseTime.Add(time.Duration(i) * time.Second),
-// 		}
-// 		err = model.DB.Create(segment).Error
-// 		require.NoError(t, err)
-// 	}
+func TestSegmentPerf(t *testing.T) {
+	config.DisableSQLLogging()
+	// dburl := filepath.Join(t.TempDir(), "test.db")
+	db, err := MakeDB(":memory:")
+	require.NoError(t, err)
+	// Create a model instance
+	model := db.(*DBModel)
+	t.Cleanup(func() {
+		// os.Remove(dburl)
+	})
 
-// 	// Verify we have 100 segments
-// 	var count int64
-// 	err = model.DB.Model(&Segment{}).Count(&count).Error
-// 	require.NoError(t, err)
-// 	require.Equal(t, int64(100), count)
+	// Create a repo for testing
+	repo := &Repo{
+		DID: "did:plc:test123",
+	}
+	err = model.DB.Create(repo).Error
+	require.NoError(t, err)
 
-// 	// Run the segment cleaner
-// 	err = model.SegmentCleaner(context.Background())
-// 	require.NoError(t, err)
+	defer config.EnableSQLLogging()
+	// Create 250000 segments with timestamps 1 hour ago, each one second apart
+	wg := sync.WaitGroup{}
+	segCount := 250000
+	wg.Add(segCount)
+	baseTime := time.Now()
+	for i := 0; i < segCount; i++ {
+		segment := &Segment{
+			ID:        fmt.Sprintf("segment-%d", i),
+			RepoDID:   repo.DID,
+			StartTime: baseTime.Add(-time.Duration(i) * time.Second).UTC(),
+		}
+		go func() {
+			defer wg.Done()
+			err = model.DB.Create(segment).Error
+			require.NoError(t, err)
+		}()
+	}
+	wg.Wait()
 
-// 	// Verify we now have only 10 segments
-// 	err = model.DB.Model(&Segment{}).Count(&count).Error
-// 	require.NoError(t, err)
-// 	require.Equal(t, int64(10), count)
-
-// 	// Verify the remaining segments are the most recent ones
-// 	var segments []Segment
-// 	err = model.DB.Model(&Segment{}).Order("start_time DESC").Find(&segments).Error
-// 	require.NoError(t, err)
-// 	require.Len(t, segments, 10)
-
-// 	// The segments should be the last 10 we created (the most recent ones)
-// 	for i, segment := range segments {
-// 		expectedTime := baseTime.Add(time.Duration(99-i) * time.Second)
-// 		// Allow a small tolerance for time comparison
-// 		require.WithinDuration(t, expectedTime, segment.StartTime, time.Second)
-// 	}
-// }
+	startTime := time.Now()
+	wg = sync.WaitGroup{}
+	runs := 1000
+	wg.Add(runs)
+	for i := 0; i < runs; i++ {
+		go func() {
+			defer wg.Done()
+			_, err := model.MostRecentSegments()
+			require.NoError(t, err)
+			// require.Len(t, segments, 1)
+		}()
+	}
+	wg.Wait()
+	fmt.Printf("Time taken: %s\n", time.Since(startTime))
+	require.Less(t, time.Since(startTime), 10*time.Second)
+}