generated from VLADIMIR/template
up
This commit is contained in:
@@ -49,7 +49,44 @@ func (r *GameResultsRepo) Capture(ctx context.Context, gameID int) (*Snapshot, e
|
||||
s.scenario,
|
||||
g.scenario_id,
|
||||
(i.mappings->>'scenario_id')::integer,
|
||||
jsonb_build_object(
|
||||
`+resultsDataSQL+`
|
||||
FROM games g
|
||||
JOIN scenarios s ON s.id = g.scenario_id
|
||||
JOIN game_imports i ON i.source_game_id = g.id AND i.state = 'ready'
|
||||
WHERE g.id = $1`,
|
||||
gameID,
|
||||
).Scan(&snapshot.ImportID, &snapshot.PackageID, &snapshot.SourceServer,
|
||||
&snapshot.SHA256, &snapshot.OriginalData, &snapshot.Manifest, &snapshot.Story, &snapshot.ScenarioID, &snapshot.ImportedScenarioID, &snapshot.Data)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrNotImported
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return snapshot, nil
|
||||
}
|
||||
|
||||
// CaptureOnline reads returned fields inside the caller's locked transaction.
|
||||
func CaptureOnline(ctx context.Context, tx pgx.Tx, gameID int) (*Snapshot, error) {
|
||||
snapshot := &Snapshot{}
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
g.scenario_id,
|
||||
s.scenario,
|
||||
`+resultsDataSQL+`
|
||||
FROM games g
|
||||
JOIN scenarios s ON s.id = g.scenario_id
|
||||
WHERE g.id = $1`,
|
||||
gameID,
|
||||
).Scan(&snapshot.ScenarioID, &snapshot.Story, &snapshot.Data)
|
||||
return snapshot, err
|
||||
}
|
||||
|
||||
const resultsDataSQL = `jsonb_build_object(
|
||||
'game', jsonb_build_object(
|
||||
'id', g.id, 'status', g.status, 'format', g.format,
|
||||
'started_at', g.started_at AT TIME ZONE 'UTC',
|
||||
@@ -131,22 +168,4 @@ func (r *GameResultsRepo) Capture(ctx context.Context, gameID int) (*Snapshot, e
|
||||
JOIN teams t ON t.id = a.team_id
|
||||
WHERE t.game_id = g.id
|
||||
), '[]'::jsonb)
|
||||
)
|
||||
FROM games g
|
||||
JOIN scenarios s ON s.id = g.scenario_id
|
||||
JOIN game_imports i ON i.source_game_id = g.id AND i.state = 'ready'
|
||||
WHERE g.id = $1`,
|
||||
gameID,
|
||||
).Scan(&snapshot.ImportID, &snapshot.PackageID, &snapshot.SourceServer,
|
||||
&snapshot.SHA256, &snapshot.OriginalData, &snapshot.Manifest, &snapshot.Story, &snapshot.ScenarioID, &snapshot.ImportedScenarioID, &snapshot.Data)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrNotImported
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return snapshot, nil
|
||||
}
|
||||
)`
|
||||
|
||||
@@ -0,0 +1,260 @@
|
||||
// Package game_returns_repo persists atomic returns and their private audit records.
|
||||
package game_returns_repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/repos/game_exports_repo"
|
||||
"evening_detective_server/internal/repos/game_results_repo"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrConflict = errors.New("пакет устарел или идентификатор уже занят")
|
||||
ErrSource = errors.New("исходный экспорт не найден")
|
||||
ErrRetry = errors.New("возврат временно недоступен; повторите попытку")
|
||||
)
|
||||
|
||||
// Return is retained even after the source game is deleted.
|
||||
type Return struct {
|
||||
PackageID string
|
||||
SHA256 string
|
||||
ExportID string
|
||||
GameID int
|
||||
PreparedAt time.Time
|
||||
AppliedAt time.Time
|
||||
Mappings []byte
|
||||
LocalActors []byte
|
||||
InitiatorID int
|
||||
}
|
||||
|
||||
type GameReturnsRepo struct{ pool *pgxpool.Pool }
|
||||
|
||||
func NewGameReturnsRepo(pool *pgxpool.Pool) *GameReturnsRepo { return &GameReturnsRepo{pool: pool} }
|
||||
|
||||
// Transaction retries only PostgreSQL serialization failures and deadlocks.
|
||||
func (r *GameReturnsRepo) Transaction(ctx context.Context, apply func(pgx.Tx) (*Return, error)) (*Return, error) {
|
||||
for attempt := 0; attempt < 3; attempt++ {
|
||||
tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := apply(tx)
|
||||
if err == nil {
|
||||
err = tx.Commit(ctx)
|
||||
}
|
||||
_ = tx.Rollback(ctx)
|
||||
if err == nil {
|
||||
return result, nil
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
var pgErr *pgconn.PgError
|
||||
if errors.As(err, &pgErr) {
|
||||
switch pgErr.Code {
|
||||
case "40001", "40P01":
|
||||
continue
|
||||
case "23505", "23503", "23514", "22001", "22003", "2200H":
|
||||
return nil, ErrConflict
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return nil, ErrRetry
|
||||
}
|
||||
|
||||
func ReadReturn(ctx context.Context, tx pgx.Tx, packageID string) (*Return, error) {
|
||||
return readReturn(ctx, tx, `WHERE package_id = $1`, packageID)
|
||||
}
|
||||
func ReadLatest(ctx context.Context, tx pgx.Tx, exportID string) (*Return, error) {
|
||||
return readReturn(ctx, tx, `WHERE export_id = $1
|
||||
ORDER BY prepared_at DESC
|
||||
LIMIT 1`, exportID)
|
||||
}
|
||||
|
||||
type IRowReader interface {
|
||||
QueryRow(context.Context, string, ...any) pgx.Row
|
||||
}
|
||||
|
||||
func (r *GameReturnsRepo) FindReturn(ctx context.Context, packageID string) (*Return, error) {
|
||||
return readReturn(ctx, r.pool, `WHERE package_id = $1`, packageID)
|
||||
}
|
||||
|
||||
func readReturn(ctx context.Context, tx IRowReader, filter, id string) (*Return, error) {
|
||||
result := &Return{}
|
||||
var prepared string
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
package_id,
|
||||
sha256,
|
||||
export_id,
|
||||
source_game_id,
|
||||
prepared_at,
|
||||
applied_at,
|
||||
mappings,
|
||||
local_actors,
|
||||
COALESCE(initiator_id, 0)
|
||||
FROM game_returns
|
||||
`+filter,
|
||||
id,
|
||||
).Scan(&result.PackageID, &result.SHA256, &result.ExportID, &result.GameID, &prepared, &result.AppliedAt, &result.Mappings, &result.LocalActors, &result.InitiatorID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result.PreparedAt, err = time.Parse(time.RFC3339Nano, prepared)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (r *GameReturnsRepo) ReadExport(ctx context.Context, packageID string) (*game_exports_repo.Attempt, error) {
|
||||
result := &game_exports_repo.Attempt{}
|
||||
err := r.pool.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
id,
|
||||
COALESCE(game_id, 0),
|
||||
package_id,
|
||||
sha256,
|
||||
size,
|
||||
source_server,
|
||||
object_key,
|
||||
snapshot
|
||||
FROM game_exports
|
||||
WHERE package_id = $1 AND state = 'ready'`,
|
||||
packageID,
|
||||
).Scan(&result.ID, &result.GameID, &result.PackageID, &result.SHA256, &result.Size, &result.SourceServer, &result.ObjectKey, &result.Snapshot)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrSource
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// Lock blocks concurrent returns, edits and new FK-linked gameplay rows for this game.
|
||||
func Lock(ctx context.Context, tx pgx.Tx, gameID int, exportID string) error {
|
||||
var id int
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
g.id
|
||||
FROM games g
|
||||
JOIN game_exports e ON e.game_id = g.id
|
||||
WHERE g.id = $1 AND e.id = $2 AND e.state = 'ready'
|
||||
FOR UPDATE OF g, e`,
|
||||
gameID,
|
||||
exportID,
|
||||
).Scan(&id)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return ErrSource
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, query := range []string{
|
||||
`SELECT
|
||||
s.id
|
||||
FROM scenarios s
|
||||
JOIN games g ON g.scenario_id = s.id
|
||||
WHERE g.id = $1
|
||||
FOR UPDATE OF s`,
|
||||
`SELECT
|
||||
id
|
||||
FROM teams
|
||||
WHERE game_id = $1
|
||||
ORDER BY id
|
||||
FOR UPDATE`,
|
||||
`SELECT
|
||||
id
|
||||
FROM tickets
|
||||
WHERE game_id = $1
|
||||
ORDER BY id
|
||||
FOR UPDATE`,
|
||||
`SELECT
|
||||
a.id
|
||||
FROM actions a
|
||||
JOIN teams t ON t.id = a.team_id
|
||||
WHERE t.game_id = $1
|
||||
ORDER BY a.id
|
||||
FOR UPDATE OF a`,
|
||||
`SELECT
|
||||
a.id
|
||||
FROM applications a
|
||||
JOIN teams t ON t.id = a.team_id
|
||||
WHERE t.game_id = $1
|
||||
ORDER BY a.id
|
||||
FOR UPDATE OF a`,
|
||||
`SELECT
|
||||
a.id
|
||||
FROM team_answers a
|
||||
JOIN teams t ON t.id = a.team_id
|
||||
WHERE t.game_id = $1
|
||||
ORDER BY a.id
|
||||
FOR UPDATE OF a`,
|
||||
} {
|
||||
rows, err := tx.Query(
|
||||
ctx,
|
||||
query,
|
||||
gameID,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for rows.Next() {
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func Capture(ctx context.Context, tx pgx.Tx, gameID int) (*game_results_repo.Snapshot, error) {
|
||||
return game_results_repo.CaptureOnline(ctx, tx, gameID)
|
||||
}
|
||||
|
||||
// Allocate reserves online identities; rolled-back sequence gaps are intentional.
|
||||
func Allocate(ctx context.Context, tx pgx.Tx, kind string) (int, error) {
|
||||
sequences := map[string]string{"tickets": "tickets_id_seq", "actions": "actions_id_seq", "applications": "applications_id_seq"}
|
||||
sequence, ok := sequences[kind]
|
||||
if !ok {
|
||||
return 0, ErrConflict
|
||||
}
|
||||
var id int
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT nextval($1::regclass)`,
|
||||
sequence,
|
||||
).Scan(&id)
|
||||
return id, err
|
||||
}
|
||||
|
||||
func Save(ctx context.Context, tx pgx.Tx, result *Return) error {
|
||||
return tx.QueryRow(
|
||||
ctx,
|
||||
`INSERT INTO game_returns (
|
||||
package_id, sha256, export_id, source_game_id, prepared_at, mappings, local_actors, initiator_id
|
||||
)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,NULLIF($8,0))
|
||||
RETURNING applied_at`,
|
||||
result.PackageID,
|
||||
result.SHA256,
|
||||
result.ExportID,
|
||||
result.GameID,
|
||||
result.PreparedAt.UTC().Format("2006-01-02T15:04:05.000000000Z"),
|
||||
result.Mappings,
|
||||
result.LocalActors,
|
||||
result.InitiatorID,
|
||||
).Scan(&result.AppliedAt)
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package game_returns_repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func TestTransactionRetriesOnlyTransientFailures(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, code string
|
||||
failures, attempts int
|
||||
expected error
|
||||
}{
|
||||
{"serialization then success", "40001", 2, 3, nil},
|
||||
{"deadlock then success", "40P01", 1, 2, nil},
|
||||
{"serialization exhausted", "40001", 3, 3, ErrRetry},
|
||||
{"deadlock exhausted", "40P01", 3, 3, ErrRetry},
|
||||
{"unique collision", "23505", 1, 1, ErrConflict},
|
||||
{"foreign key collision", "23503", 1, 1, ErrConflict},
|
||||
{"sequence exhausted", "2200H", 1, 1, ErrConflict},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("set TEST_DATABASE_URL for transaction retry checks")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer pool.Close()
|
||||
repo := NewGameReturnsRepo(pool)
|
||||
attempts := 0
|
||||
result, err := repo.Transaction(ctx, func(pgx.Tx) (*Return, error) {
|
||||
attempts++
|
||||
if attempts <= tc.failures {
|
||||
return nil, &pgconn.PgError{Code: tc.code}
|
||||
}
|
||||
return &Return{PackageID: "saved"}, nil
|
||||
})
|
||||
if attempts != tc.attempts || !errors.Is(err, tc.expected) {
|
||||
t.Fatalf("attempts %d error %v", attempts, err)
|
||||
}
|
||||
if err == nil && result.PackageID != "saved" {
|
||||
t.Fatal("result lost")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSequenceErrorPreservesCancellationAndCause(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
cancelParent, cancelAdvance bool
|
||||
expected error
|
||||
}{
|
||||
{"parent cancelled", true, true, context.Canceled},
|
||||
{"advance cancelled", false, true, ErrRetry},
|
||||
{"driver failure", false, false, ErrConflict},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
advanceCtx, advanceCancel := context.WithCancel(ctx)
|
||||
defer advanceCancel()
|
||||
if tc.cancelParent {
|
||||
cancel()
|
||||
}
|
||||
if tc.cancelAdvance {
|
||||
advanceCancel()
|
||||
}
|
||||
err := sequenceError(ctx, advanceCtx, ErrConflict)
|
||||
if !errors.Is(err, tc.expected) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,304 @@
|
||||
package game_returns_repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// Write replaces gameplay only; technical mail fields and existing online actors survive updates.
|
||||
func Write(ctx context.Context, tx pgx.Tx, gameID int, data []byte) error {
|
||||
var collision bool
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM teams t
|
||||
JOIN jsonb_to_recordset($2::jsonb->'teams') AS n(id integer) ON n.id = t.id
|
||||
WHERE t.game_id <> $1
|
||||
) OR EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM tickets t
|
||||
JOIN jsonb_to_recordset($2::jsonb->'tickets') AS n(id integer) ON n.id = t.id
|
||||
WHERE t.game_id <> $1
|
||||
)`,
|
||||
gameID,
|
||||
data,
|
||||
).Scan(&collision)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if collision {
|
||||
return ErrConflict
|
||||
}
|
||||
for _, query := range []string{
|
||||
`DELETE FROM actions a
|
||||
USING teams t
|
||||
WHERE a.team_id = t.id AND t.game_id = $1 AND $2::jsonb IS NOT NULL`,
|
||||
`DELETE FROM applications a
|
||||
USING teams t
|
||||
WHERE a.team_id = t.id AND t.game_id = $1 AND $2::jsonb IS NOT NULL`,
|
||||
`DELETE FROM team_answers a
|
||||
USING teams t
|
||||
WHERE a.team_id = t.id AND t.game_id = $1 AND $2::jsonb IS NOT NULL`,
|
||||
`DELETE FROM game_map_items m
|
||||
WHERE m.game_id = $1 AND m.kind = 'team'
|
||||
AND NOT EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM jsonb_to_recordset($2::jsonb->'teams') AS n(id integer)
|
||||
WHERE n.id = m.item_id
|
||||
)`,
|
||||
`DELETE FROM teams t
|
||||
WHERE t.game_id = $1
|
||||
AND NOT EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM jsonb_to_recordset($2::jsonb->'teams') AS n(id integer)
|
||||
WHERE n.id = t.id
|
||||
)`,
|
||||
`DELETE FROM tickets t
|
||||
WHERE t.game_id = $1
|
||||
AND NOT EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM jsonb_to_recordset($2::jsonb->'tickets') AS n(id integer)
|
||||
WHERE n.id = t.id
|
||||
)`,
|
||||
`INSERT INTO teams (id, game_id, name, status, password, creator_id, captain_email, participants_count, participation_status, participation_source, participation_answered_at, registration_delivery_id, created_at)
|
||||
SELECT
|
||||
n.id,
|
||||
n.game_id,
|
||||
n.name,
|
||||
n.status,
|
||||
n.password,
|
||||
CASE WHEN EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM users u
|
||||
WHERE u.id = n.creator_id
|
||||
) THEN n.creator_id ELSE NULL END,
|
||||
n.captain_email,
|
||||
n.participants_count,
|
||||
n.participation_status,
|
||||
n.participation_source,
|
||||
n.participation_answered_at,
|
||||
n.registration_delivery_id,
|
||||
n.created_at
|
||||
FROM jsonb_to_recordset($2::jsonb->'teams') AS n(
|
||||
id integer,
|
||||
game_id integer,
|
||||
name text,
|
||||
status text,
|
||||
password text,
|
||||
creator_id integer,
|
||||
captain_email text,
|
||||
participants_count integer,
|
||||
participation_status text,
|
||||
participation_source text,
|
||||
participation_answered_at timestamptz,
|
||||
registration_delivery_id text,
|
||||
created_at timestamp
|
||||
)
|
||||
ON CONFLICT (id) DO UPDATE
|
||||
SET name = EXCLUDED.name,
|
||||
status = EXCLUDED.status,
|
||||
password = EXCLUDED.password,
|
||||
captain_email = EXCLUDED.captain_email,
|
||||
participants_count = EXCLUDED.participants_count,
|
||||
participation_status = EXCLUDED.participation_status,
|
||||
participation_source = EXCLUDED.participation_source,
|
||||
participation_answered_at = EXCLUDED.participation_answered_at,
|
||||
created_at = EXCLUDED.created_at
|
||||
WHERE teams.game_id = $1`,
|
||||
`INSERT INTO tickets (id, game_id, token_hash, status, used_at, used_by, created_at)
|
||||
SELECT
|
||||
n.id,
|
||||
n.game_id,
|
||||
decode(n.token_hash, 'base64'),
|
||||
n.status,
|
||||
n.used_at,
|
||||
CASE WHEN EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM users u
|
||||
WHERE u.id = n.used_by
|
||||
) THEN n.used_by ELSE NULL END,
|
||||
n.created_at
|
||||
FROM jsonb_to_recordset($2::jsonb->'tickets') AS n(
|
||||
id integer,
|
||||
game_id integer,
|
||||
token_hash text,
|
||||
status text,
|
||||
used_at timestamptz,
|
||||
used_by integer,
|
||||
created_at timestamptz
|
||||
)
|
||||
ON CONFLICT (id) DO UPDATE
|
||||
SET token_hash = EXCLUDED.token_hash,
|
||||
status = EXCLUDED.status,
|
||||
used_at = EXCLUDED.used_at,
|
||||
used_by = EXCLUDED.used_by,
|
||||
created_at = EXCLUDED.created_at
|
||||
WHERE tickets.game_id = $1`,
|
||||
`INSERT INTO actions (id, team_id, code, via_mcp, created_at)
|
||||
SELECT
|
||||
n.id,
|
||||
n.team_id,
|
||||
n.code,
|
||||
n.via_mcp,
|
||||
n.created_at
|
||||
FROM jsonb_to_recordset($2::jsonb->'actions') AS n(
|
||||
id integer,
|
||||
team_id integer,
|
||||
code text,
|
||||
via_mcp boolean,
|
||||
created_at timestamp
|
||||
)
|
||||
WHERE EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM teams t
|
||||
WHERE t.id = n.team_id AND t.game_id = $1
|
||||
)`,
|
||||
`INSERT INTO applications (id, team_id, name, image, file_type, status, created_at)
|
||||
SELECT
|
||||
n.id,
|
||||
n.team_id,
|
||||
n.name,
|
||||
n.image,
|
||||
n.file_type,
|
||||
n.status,
|
||||
n.created_at
|
||||
FROM jsonb_to_recordset($2::jsonb->'applications') AS n(
|
||||
id integer,
|
||||
team_id integer,
|
||||
name text,
|
||||
image text,
|
||||
file_type text,
|
||||
status text,
|
||||
created_at timestamp
|
||||
)
|
||||
WHERE EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM teams t
|
||||
WHERE t.id = n.team_id AND t.game_id = $1
|
||||
)`,
|
||||
`INSERT INTO team_answers (team_id, question_code, answer, score, created_at, updated_at)
|
||||
SELECT
|
||||
n.team_id,
|
||||
n.question_code,
|
||||
n.answer,
|
||||
n.score,
|
||||
n.created_at,
|
||||
n.updated_at
|
||||
FROM jsonb_to_recordset($2::jsonb->'answers') AS n(
|
||||
team_id integer,
|
||||
question_code text,
|
||||
answer text,
|
||||
score integer,
|
||||
created_at timestamp,
|
||||
updated_at timestamp
|
||||
)
|
||||
WHERE EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM teams t
|
||||
WHERE t.id = n.team_id AND t.game_id = $1
|
||||
)`,
|
||||
`UPDATE games
|
||||
SET format = $2::jsonb->'game'->>'format',
|
||||
status = $2::jsonb->'game'->>'status',
|
||||
started_at = ($2::jsonb->'game'->>'started_at')::timestamp,
|
||||
ended_at = ($2::jsonb->'game'->>'ended_at')::timestamp
|
||||
WHERE id = $1`,
|
||||
} {
|
||||
if _, err := tx.Exec(
|
||||
ctx,
|
||||
query,
|
||||
gameID,
|
||||
data,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// nextval is monotonic under concurrent inserts; setval can move a live sequence backwards.
|
||||
// shortcut: huge venue ID gaps consume sequence values even on refusal; widen IDs before supporting such venues.
|
||||
var maximum int
|
||||
if err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT COALESCE(max(id),0)
|
||||
FROM teams`,
|
||||
).Scan(&maximum); err != nil {
|
||||
return err
|
||||
}
|
||||
if maximum >= 2147483647 {
|
||||
return ErrConflict
|
||||
}
|
||||
advanceCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
for {
|
||||
var id int
|
||||
if err := tx.QueryRow(
|
||||
advanceCtx,
|
||||
`SELECT nextval('teams_id_seq')`,
|
||||
).Scan(&id); err != nil {
|
||||
return sequenceError(ctx, advanceCtx, err)
|
||||
}
|
||||
if id >= maximum {
|
||||
break
|
||||
}
|
||||
if err := tx.QueryRow(
|
||||
advanceCtx,
|
||||
`SELECT max(nextval('teams_id_seq'))
|
||||
FROM generate_series(1,$1::integer)`,
|
||||
min(maximum-id, 4096),
|
||||
).Scan(&id); err != nil {
|
||||
return sequenceError(ctx, advanceCtx, err)
|
||||
}
|
||||
if id >= maximum {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func sequenceError(ctx, advanceCtx context.Context, err error) error {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
if advanceCtx.Err() != nil {
|
||||
return ErrRetry
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// RestoreStory isolates returned questions and materials from other games sharing the scenario.
|
||||
func RestoreStory(ctx context.Context, tx pgx.Tx, gameID int, story []byte) (int, error) {
|
||||
var scenarioID int
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`INSERT INTO scenarios (name, description, image, scenario, author_id, status, playlist_id)
|
||||
SELECT s.name, s.description, s.image, $2::jsonb, s.author_id, 'draft', s.playlist_id
|
||||
FROM scenarios s
|
||||
JOIN games g ON g.scenario_id = s.id
|
||||
WHERE g.id = $1
|
||||
RETURNING id`,
|
||||
gameID,
|
||||
story,
|
||||
).Scan(&scenarioID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
_, err = tx.Exec(
|
||||
ctx,
|
||||
`UPDATE games SET scenario_id = $2 WHERE id = $1`,
|
||||
gameID,
|
||||
scenarioID,
|
||||
)
|
||||
return scenarioID, err
|
||||
}
|
||||
@@ -78,10 +78,14 @@ func (r *GamesRepo) GetGameByID(
|
||||
games.started_at,
|
||||
games.ended_at,
|
||||
(SELECT
|
||||
jsonb_build_object(
|
||||
CASE WHEN EXISTS (
|
||||
SELECT 1
|
||||
FROM game_returns r
|
||||
WHERE r.export_id = e.id
|
||||
) THEN NULL ELSE jsonb_build_object(
|
||||
'id', e.id, 'state', e.state, 'package_id', e.package_id,
|
||||
'size', e.size, 'sha256', e.sha256, 'failure', e.failure
|
||||
)
|
||||
) END
|
||||
FROM game_exports e
|
||||
WHERE e.game_id = games.id
|
||||
ORDER BY e.created_at DESC, e.id
|
||||
@@ -148,10 +152,14 @@ func (r *GamesRepo) GetGames(
|
||||
games.format,
|
||||
games.status,
|
||||
(SELECT
|
||||
jsonb_build_object(
|
||||
CASE WHEN EXISTS (
|
||||
SELECT 1
|
||||
FROM game_returns r
|
||||
WHERE r.export_id = e.id
|
||||
) THEN NULL ELSE jsonb_build_object(
|
||||
'id', e.id, 'state', e.state, 'package_id', e.package_id,
|
||||
'size', e.size, 'sha256', e.sha256, 'failure', e.failure
|
||||
)
|
||||
) END
|
||||
FROM game_exports e
|
||||
WHERE e.game_id = games.id
|
||||
ORDER BY e.created_at DESC, e.id
|
||||
|
||||
@@ -162,20 +162,56 @@ func (r *TeamsRepo) FinishParticipationEmail(ctx context.Context, id, attempt in
|
||||
// ClaimReviewReminder claims the next review reminder job.
|
||||
func (r *TeamsRepo) ClaimReviewReminder(ctx context.Context) (*ReviewReminder, error) {
|
||||
job := &ReviewReminder{}
|
||||
err := r.pool.QueryRow(ctx, `WITH candidate AS (
|
||||
SELECT t.id FROM teams t JOIN games g ON g.id=t.game_id
|
||||
WHERE t.captain_email<>'' AND t.review_reminder_sent_at IS NULL
|
||||
AND t.review_reminder_attempts < $1
|
||||
AND (t.review_reminder_next_attempt_at IS NULL OR t.review_reminder_next_attempt_at <= NOW())
|
||||
AND g.status='finish' AND g.started_at <= NOW() - INTERVAL '16 hours'
|
||||
AND NOT EXISTS (SELECT 1 FROM game_reviews r WHERE r.game_id=t.game_id AND r.team_id=t.id)
|
||||
ORDER BY g.started_at, t.id
|
||||
FOR UPDATE OF t SKIP LOCKED LIMIT 1
|
||||
) UPDATE teams t SET review_reminder_next_attempt_at=NOW()+INTERVAL '5 minutes',
|
||||
review_reminder_attempts=t.review_reminder_attempts+1
|
||||
FROM candidate c, games g WHERE t.id=c.id AND g.id=t.game_id
|
||||
RETURNING t.id,t.name,t.game_id,g.name,t.captain_email,t.password,t.review_reminder_attempts`, maxReviewReminderAttempts).Scan(
|
||||
&job.TeamID, &job.TeamName, &job.GameID, &job.GameName, &job.Email, &job.Password, &job.Attempts)
|
||||
err := r.pool.QueryRow(
|
||||
ctx,
|
||||
`WITH candidate AS (
|
||||
SELECT
|
||||
t.id
|
||||
FROM teams t
|
||||
JOIN games g ON g.id = t.game_id
|
||||
WHERE t.captain_email <> '' AND t.review_reminder_sent_at IS NULL
|
||||
AND t.review_reminder_attempts < $1
|
||||
AND (t.review_reminder_next_attempt_at IS NULL OR t.review_reminder_next_attempt_at <= NOW())
|
||||
AND g.status = 'finish' AND g.started_at <= NOW() - INTERVAL '16 hours'
|
||||
AND NOT EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM game_reviews r
|
||||
WHERE r.game_id = t.game_id AND r.team_id = t.id
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM game_returns r
|
||||
WHERE r.source_game_id = g.id
|
||||
)
|
||||
ORDER BY g.started_at, t.id
|
||||
FOR UPDATE OF t SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
UPDATE teams t
|
||||
SET review_reminder_next_attempt_at = NOW() + INTERVAL '5 minutes',
|
||||
review_reminder_attempts = t.review_reminder_attempts + 1
|
||||
FROM candidate c, games g
|
||||
WHERE t.id = c.id AND g.id = t.game_id
|
||||
RETURNING
|
||||
t.id,
|
||||
t.name,
|
||||
t.game_id,
|
||||
g.name,
|
||||
t.captain_email,
|
||||
t.password,
|
||||
t.review_reminder_attempts`,
|
||||
maxReviewReminderAttempts,
|
||||
).Scan(
|
||||
&job.TeamID,
|
||||
&job.TeamName,
|
||||
&job.GameID,
|
||||
&job.GameName,
|
||||
&job.Email,
|
||||
&job.Password,
|
||||
&job.Attempts,
|
||||
)
|
||||
return job, err
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user