feat(playsessions): add FindOrCreate session service (30-min rolling window)
Per spec §6: returns existing session id when last_event_at is within timeout; inserts a new session otherwise. Touch updates last_event_at and increments track_count so callers can assume FindOrCreate == one play started.
This commit is contained in:
@@ -0,0 +1,72 @@
|
||||
// Package playsessions implements the listening-session lifecycle from
|
||||
// spec §6: a session is a contiguous sequence of plays by a user with no
|
||||
// gap longer than the configured timeout (default 30 minutes) between
|
||||
// events.
|
||||
package playsessions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
// FindOrCreate returns the session id for the given user and event time.
|
||||
// If the user's most recent session was last touched within `timeout`, that
|
||||
// session is extended (last_event_at and track_count updated). Otherwise a
|
||||
// new session is inserted.
|
||||
//
|
||||
// q is *dbq.Queries; pass either the pool-level Queries or one bound to a
|
||||
// transaction so callers can compose this into a larger atomic operation.
|
||||
func FindOrCreate(
|
||||
ctx context.Context,
|
||||
q *dbq.Queries,
|
||||
userID pgtype.UUID,
|
||||
eventTime time.Time,
|
||||
clientID string,
|
||||
timeout time.Duration,
|
||||
) (pgtype.UUID, error) {
|
||||
prev, err := q.GetMostRecentPlaySessionForUser(ctx, userID)
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return pgtype.UUID{}, err
|
||||
}
|
||||
if err == nil {
|
||||
// "Within window" includes the exact-boundary case: gap <= timeout extends.
|
||||
gap := eventTime.Sub(prev.LastEventAt.Time)
|
||||
if gap <= timeout {
|
||||
if err := q.TouchPlaySessionLastEvent(ctx, dbq.TouchPlaySessionLastEventParams{
|
||||
ID: prev.ID,
|
||||
LastEventAt: pgtype.Timestamptz{Time: eventTime, Valid: true},
|
||||
}); err != nil {
|
||||
return pgtype.UUID{}, err
|
||||
}
|
||||
return prev.ID, nil
|
||||
}
|
||||
}
|
||||
// Either no prior session, or the gap exceeded the timeout.
|
||||
var clientIDPtr *string
|
||||
if clientID != "" {
|
||||
clientIDPtr = &clientID
|
||||
}
|
||||
row, err := q.InsertPlaySession(ctx, dbq.InsertPlaySessionParams{
|
||||
UserID: userID,
|
||||
StartedAt: pgtype.Timestamptz{Time: eventTime, Valid: true},
|
||||
ClientID: clientIDPtr,
|
||||
})
|
||||
if err != nil {
|
||||
return pgtype.UUID{}, err
|
||||
}
|
||||
// Bump track_count to 1 on creation so the contract "every FindOrCreate is
|
||||
// the start of one play" stays true (insert leaves track_count default 0).
|
||||
if err := q.TouchPlaySessionLastEvent(ctx, dbq.TouchPlaySessionLastEventParams{
|
||||
ID: row.ID,
|
||||
LastEventAt: pgtype.Timestamptz{Time: eventTime, Valid: true},
|
||||
}); err != nil {
|
||||
return pgtype.UUID{}, err
|
||||
}
|
||||
return row.ID, nil
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package playsessions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
func testPool(t *testing.T) *pgxpool.Pool {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test in -short mode")
|
||||
}
|
||||
dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("MINSTREL_TEST_DATABASE_URL not set")
|
||||
}
|
||||
if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
pool, err := pgxpool.New(context.Background(), dsn)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
t.Cleanup(pool.Close)
|
||||
if _, err := pool.Exec(context.Background(),
|
||||
"TRUNCATE play_events, skip_events, play_sessions, sessions, users RESTART IDENTITY CASCADE"); err != nil {
|
||||
t.Fatalf("truncate: %v", err)
|
||||
}
|
||||
return pool
|
||||
}
|
||||
|
||||
func seedTestUser(t *testing.T, pool *pgxpool.Pool) pgtype.UUID {
|
||||
t.Helper()
|
||||
u, err := dbq.New(pool).CreateUser(context.Background(), dbq.CreateUserParams{
|
||||
Username: "tester",
|
||||
PasswordHash: "x",
|
||||
ApiToken: "x",
|
||||
IsAdmin: false,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateUser: %v", err)
|
||||
}
|
||||
return u.ID
|
||||
}
|
||||
|
||||
func TestFindOrCreate_NoPriorSessionCreatesOne(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
user := seedTestUser(t, pool)
|
||||
q := dbq.New(pool)
|
||||
|
||||
now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC)
|
||||
id, err := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("FindOrCreate: %v", err)
|
||||
}
|
||||
if !id.Valid {
|
||||
t.Fatalf("session id not valid")
|
||||
}
|
||||
row, err := q.GetMostRecentPlaySessionForUser(context.Background(), user)
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
if row.ID != id {
|
||||
t.Errorf("returned id %v, db has %v", id, row.ID)
|
||||
}
|
||||
if !row.StartedAt.Time.Equal(now) {
|
||||
t.Errorf("started_at = %v, want %v", row.StartedAt.Time, now)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindOrCreate_WithinWindowExtendsExisting(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
user := seedTestUser(t, pool)
|
||||
q := dbq.New(pool)
|
||||
|
||||
now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC)
|
||||
first, err := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("first: %v", err)
|
||||
}
|
||||
second, err := FindOrCreate(context.Background(), q, user, now.Add(15*time.Minute), "client-a", 30*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("second: %v", err)
|
||||
}
|
||||
if first != second {
|
||||
t.Errorf("session id changed: %v -> %v", first, second)
|
||||
}
|
||||
row, err := q.GetMostRecentPlaySessionForUser(context.Background(), user)
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
if !row.LastEventAt.Time.Equal(now.Add(15 * time.Minute)) {
|
||||
t.Errorf("last_event_at = %v", row.LastEventAt.Time)
|
||||
}
|
||||
if row.TrackCount != 2 {
|
||||
t.Errorf("track_count = %d, want 2", row.TrackCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindOrCreate_BeyondWindowStartsNew(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
user := seedTestUser(t, pool)
|
||||
q := dbq.New(pool)
|
||||
|
||||
now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC)
|
||||
first, err := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("first: %v", err)
|
||||
}
|
||||
second, err := FindOrCreate(context.Background(), q, user, now.Add(31*time.Minute), "client-a", 30*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("second: %v", err)
|
||||
}
|
||||
if first == second {
|
||||
t.Errorf("expected new session, got same id %v", first)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindOrCreate_AtExactBoundaryExtendsExisting(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
user := seedTestUser(t, pool)
|
||||
q := dbq.New(pool)
|
||||
|
||||
now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC)
|
||||
first, _ := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute)
|
||||
second, err := FindOrCreate(context.Background(), q, user, now.Add(30*time.Minute), "client-a", 30*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("second: %v", err)
|
||||
}
|
||||
if first != second {
|
||||
t.Errorf("expected same session at exact boundary, got new id %v", second)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user