generated from VLADIMIR/template
up
This commit is contained in:
@@ -0,0 +1,97 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"evening_detective_server/internal/modules/processor_jwt"
|
||||
"evening_detective_server/internal/modules/roles"
|
||||
"evening_detective_server/internal/services/game_export_service"
|
||||
"evening_detective_server/proto"
|
||||
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func canExportGame(claims *processor_jwt.JWTClaims) bool {
|
||||
return claims != nil && (roles.HasRole(claims, roles.Organizer) || claims.HasRole(roles.Developer))
|
||||
}
|
||||
|
||||
func (s *server) StartGameExport(ctx context.Context, req *proto.StartGameExportReq) (*proto.GameExportRsp, error) {
|
||||
claims, err := claimsFromContext(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !canExportGame(claims) {
|
||||
return nil, status.Error(codes.PermissionDenied, "permission denied")
|
||||
}
|
||||
result, err := s.gameExportService.Start(ctx, int(req.GetID()), claims.UserID)
|
||||
if err != nil {
|
||||
return nil, gameExportErrorStatus(err)
|
||||
}
|
||||
return &proto.GameExportRsp{Export: mapGameExport(result)}, nil
|
||||
}
|
||||
|
||||
func (s *server) GetGameExport(ctx context.Context, req *proto.GetGameExportReq) (*proto.GameExportRsp, error) {
|
||||
claims, err := claimsFromContext(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !canExportGame(claims) {
|
||||
return nil, status.Error(codes.PermissionDenied, "permission denied")
|
||||
}
|
||||
result, err := s.gameExportService.Get(ctx, int(req.GetID()))
|
||||
if err != nil {
|
||||
return nil, gameExportErrorStatus(err)
|
||||
}
|
||||
return &proto.GameExportRsp{Export: mapGameExport(result)}, nil
|
||||
}
|
||||
|
||||
// GameExportDownloadHandler authenticates with the existing JWT processor and streams private ZIPs.
|
||||
func GameExportDownloadHandler(jwt processor_jwt.IProcessorJWT, exports *game_export_service.GameExportService) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
fail := func(err error) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(runtime.HTTPStatusFromCode(status.Code(err)))
|
||||
// The message has already been reduced to a safe transport error.
|
||||
_, _ = w.Write([]byte(`{"error":` + strconv.Quote(status.Convert(err).Message()) + `}`))
|
||||
}
|
||||
token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
|
||||
if token == "" {
|
||||
fail(status.Error(codes.Unauthenticated, "Требуется авторизация"))
|
||||
return
|
||||
}
|
||||
claims, err := jwt.GetClaims(token)
|
||||
if err != nil {
|
||||
fail(status.Error(codes.Unauthenticated, "Недействительный токен"))
|
||||
return
|
||||
}
|
||||
if !canExportGame(claims) {
|
||||
fail(status.Error(codes.PermissionDenied, "permission denied"))
|
||||
return
|
||||
}
|
||||
gameID, err := strconv.ParseInt(r.PathValue("id"), 10, 32)
|
||||
if err != nil || gameID <= 0 {
|
||||
fail(status.Error(codes.InvalidArgument, "Некорректный идентификатор игры"))
|
||||
return
|
||||
}
|
||||
result, body, err := exports.OpenDownload(r.Context(), int(gameID))
|
||||
if err != nil {
|
||||
fail(gameExportErrorStatus(err))
|
||||
return
|
||||
}
|
||||
defer func() { _ = body.Close() }()
|
||||
w.Header().Set("Content-Type", "application/zip")
|
||||
w.Header().Set("Content-Disposition", `attachment; filename="game-`+result.PackageID+`.zip"`)
|
||||
w.Header().Set("Content-Length", strconv.FormatInt(result.Size, 10))
|
||||
w.Header().Set("X-Content-Type-Options", "nosniff")
|
||||
if _, err := io.CopyBuffer(w, body, make([]byte, 32<<10)); err != nil && r.Context().Err() == nil {
|
||||
log.Printf("game export download: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"log"
|
||||
|
||||
"evening_detective_server/internal/services/game_export_service"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func gameExportErrorStatus(err error) error {
|
||||
switch {
|
||||
case errors.Is(err, game_export_service.ErrGameNotFound):
|
||||
return status.Error(codes.NotFound, game_export_service.ErrGameNotFound.Error())
|
||||
case errors.Is(err, game_export_service.ErrExportNotFound):
|
||||
return status.Error(codes.NotFound, game_export_service.ErrExportNotFound.Error())
|
||||
case errors.Is(err, game_export_service.ErrVenueMode):
|
||||
return status.Error(codes.FailedPrecondition, game_export_service.ErrVenueMode.Error())
|
||||
case errors.Is(err, game_export_service.ErrNotReady):
|
||||
return status.Error(codes.FailedPrecondition, game_export_service.ErrNotReady.Error())
|
||||
case errors.Is(err, game_export_service.ErrNotExportable):
|
||||
return status.Error(codes.FailedPrecondition, game_export_service.ErrNotExportable.Error())
|
||||
case errors.Is(err, game_export_service.ErrInvalidGameID):
|
||||
return status.Error(codes.InvalidArgument, game_export_service.ErrInvalidGameID.Error())
|
||||
default:
|
||||
log.Printf("game export API: %v", err)
|
||||
return status.Error(codes.Internal, "Не удалось выполнить операцию экспорта игры")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"evening_detective_server/internal/services/game_export_service"
|
||||
"evening_detective_server/proto"
|
||||
)
|
||||
|
||||
func mapGameExport(value *game_export_service.GameExport) *proto.GameExport {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
return &proto.GameExport{
|
||||
ID: value.ID,
|
||||
State: value.State,
|
||||
PackageID: value.PackageID,
|
||||
Size: value.Size,
|
||||
Sha256: value.SHA256,
|
||||
Failure: value.Failure,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"evening_detective_server/internal/modules/processor_jwt"
|
||||
"evening_detective_server/internal/services/game_export_service"
|
||||
"evening_detective_server/proto"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func TestExportRolesAndVenueMode(t *testing.T) {
|
||||
exports := game_export_service.NewGameExportService(nil, nil, nil, nil, "", true)
|
||||
server := &server{gameExportService: exports}
|
||||
jwt := processor_jwt.NewProcessor("export-test-secret")
|
||||
for _, tc := range []struct {
|
||||
role string
|
||||
code codes.Code
|
||||
httpCode int
|
||||
}{
|
||||
{"admin", codes.FailedPrecondition, http.StatusBadRequest},
|
||||
{"developer", codes.FailedPrecondition, http.StatusBadRequest},
|
||||
{"organizer", codes.FailedPrecondition, http.StatusBadRequest},
|
||||
{"author", codes.PermissionDenied, http.StatusForbidden},
|
||||
{"user", codes.PermissionDenied, http.StatusForbidden},
|
||||
{"", codes.Unauthenticated, http.StatusUnauthorized},
|
||||
} {
|
||||
t.Run(tc.role, func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
if tc.role != "" {
|
||||
ctx = processor_jwt.ContextWithClaims(ctx, &processor_jwt.JWTClaims{UserID: 1, Roles: []string{tc.role}})
|
||||
}
|
||||
_, err := server.StartGameExport(ctx, &proto.StartGameExportReq{ID: 1})
|
||||
if status.Code(err) != tc.code {
|
||||
t.Fatalf("start access: %v", err)
|
||||
}
|
||||
_, err = server.GetGameExport(ctx, &proto.GetGameExportReq{ID: 1})
|
||||
if status.Code(err) != tc.code {
|
||||
t.Fatalf("status access: %v", err)
|
||||
}
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/games/1/export/download", nil)
|
||||
request.SetPathValue("id", "1")
|
||||
if tc.role != "" {
|
||||
token, err := jwt.GenerateAccessToken(1, "export@test.invalid", []string{tc.role})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
request.Header.Set("Authorization", "Bearer "+token)
|
||||
}
|
||||
response := httptest.NewRecorder()
|
||||
GameExportDownloadHandler(jwt, exports).ServeHTTP(response, request)
|
||||
if response.Code != tc.httpCode {
|
||||
t.Fatalf("download access: %d, %s", response.Code, response.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExportErrorsAreSafe(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
err error
|
||||
code codes.Code
|
||||
}{
|
||||
{"game missing", game_export_service.ErrGameNotFound, codes.NotFound},
|
||||
{"export missing", game_export_service.ErrExportNotFound, codes.NotFound},
|
||||
{"progress", game_export_service.ErrNotExportable, codes.FailedPrecondition},
|
||||
{"not ready", game_export_service.ErrNotReady, codes.FailedPrecondition},
|
||||
{"venue", game_export_service.ErrVenueMode, codes.FailedPrecondition},
|
||||
{"invalid game", game_export_service.ErrInvalidGameID, codes.InvalidArgument},
|
||||
{"internal", errors.New("private SQL and object key"), codes.Internal},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
mapped := gameExportErrorStatus(fmt.Errorf("private detail: %w", tc.err))
|
||||
if status.Code(mapped) != tc.code || strings.Contains(mapped.Error(), "private") {
|
||||
t.Fatalf("unexpected export status: %v", mapped)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -29,6 +29,7 @@ func mapGame(
|
||||
}
|
||||
return &proto.Game{
|
||||
ID: int32(o.ID),
|
||||
Export: mapGameExport(o.Export),
|
||||
Name: o.Name,
|
||||
Description: o.Description,
|
||||
Location: location,
|
||||
|
||||
@@ -3,6 +3,7 @@ package app
|
||||
import (
|
||||
"evening_detective_server/internal/services/answer_service"
|
||||
"evening_detective_server/internal/services/file_service"
|
||||
"evening_detective_server/internal/services/game_export_service"
|
||||
"evening_detective_server/internal/services/game_review_service"
|
||||
"evening_detective_server/internal/services/game_service"
|
||||
"evening_detective_server/internal/services/gameplay_service"
|
||||
@@ -28,6 +29,7 @@ type server struct {
|
||||
taskService *task_service.TaskService
|
||||
fileService *file_service.FileService
|
||||
scenarioService *scenarios_service.ScenarioService
|
||||
gameExportService *game_export_service.GameExportService
|
||||
gameService *game_service.GameService
|
||||
gameplayService *gameplay_service.GameplayService
|
||||
answerService *answer_service.AnswerService
|
||||
@@ -59,9 +61,11 @@ func NewServer(
|
||||
screenPlanService *screen_plan_service.ScreenPlanService,
|
||||
mapService *map_service.MapService,
|
||||
taskService *task_service.TaskService,
|
||||
gameExportService *game_export_service.GameExportService,
|
||||
) proto.EveningDetectiveServerServer {
|
||||
return &server{
|
||||
taskService: taskService,
|
||||
gameExportService: gameExportService,
|
||||
usersService: usersService,
|
||||
uiService: uiService,
|
||||
fileService: fileService,
|
||||
|
||||
@@ -38,13 +38,19 @@ func NewRustFSStorage(
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
defer cancel()
|
||||
err = client.CreateBucket(
|
||||
ctx,
|
||||
bucket,
|
||||
)
|
||||
exists, err := client.BucketExists(ctx, bucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !exists {
|
||||
err = client.CreateBucket(ctx, bucket)
|
||||
if err != nil {
|
||||
var apiErr smithy.APIError
|
||||
if !errors.As(err, &apiErr) || apiErr.ErrorCode() != "BucketAlreadyOwnedByYou" {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return &storage{
|
||||
client: client,
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
package file_storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
)
|
||||
|
||||
// IStreamStorage supports large private objects without loading them into memory.
|
||||
type IStreamStorage interface {
|
||||
PutStream(context.Context, string, io.Reader, string) error
|
||||
Open(context.Context, string) (io.ReadCloser, error)
|
||||
Delete(context.Context, string) error
|
||||
}
|
||||
|
||||
// NewPrivateRustFSStorage uses an independent bucket with no public file route or policy.
|
||||
func NewPrivateRustFSStorage(ctx context.Context, endpoint, user, password, bucket string) (IStreamStorage, error) {
|
||||
files, err := NewRustFSStorage(ctx, endpoint, user, password, bucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return files.(*storage), nil
|
||||
}
|
||||
|
||||
func (s *storage) PutStream(ctx context.Context, key string, body io.Reader, mime string) error {
|
||||
return s.client.PutObject(ctx, s.bucket, key, body, mime)
|
||||
}
|
||||
|
||||
func (s *storage) Open(ctx context.Context, key string) (io.ReadCloser, error) {
|
||||
return s.client.GetObject(ctx, s.bucket, key)
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
|
||||
// Game describes a game and its scenario and teams.
|
||||
type Game struct {
|
||||
Export *GameExport
|
||||
Format string
|
||||
ID int
|
||||
Name string
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
package repos
|
||||
|
||||
// GameExport contains public status, never the snapshot or storage key.
|
||||
type GameExport struct {
|
||||
ID string `json:"id"`
|
||||
State string `json:"state"`
|
||||
PackageID string `json:"package_id"`
|
||||
Size int64 `json:"size"`
|
||||
SHA256 string `json:"sha256"`
|
||||
Failure string `json:"failure"`
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package game_exports_repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/repos/games_repo"
|
||||
"evening_detective_server/internal/repos/scenarios_repo"
|
||||
"evening_detective_server/internal/repos/teams_repo"
|
||||
"evening_detective_server/internal/repos/user_agreements_repo"
|
||||
"evening_detective_server/internal/repos/users_repo"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type exportFixture struct {
|
||||
pool *pgxpool.Pool
|
||||
exports *GameExportsRepo
|
||||
games *games_repo.GamesRepo
|
||||
teams *teams_repo.TeamsRepo
|
||||
scenarios *scenarios_repo.ScenariosRepo
|
||||
users *users_repo.UsersRepo
|
||||
userID int
|
||||
scenarioID int
|
||||
gameID int
|
||||
teamID int
|
||||
}
|
||||
|
||||
// newExportFixture uses real migrations in an isolated test schema; all game data uses repositories.
|
||||
func newExportFixture(t *testing.T) exportFixture {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("set TEST_DATABASE_URL for PostgreSQL export checks")
|
||||
}
|
||||
ctx := context.Background()
|
||||
admin, err := pgxpool.New(ctx, dsn)
|
||||
requireNoError(t, err)
|
||||
schema := fmt.Sprintf("exports_test_%d", time.Now().UnixNano())
|
||||
quoted := pgx.Identifier{schema}.Sanitize()
|
||||
_, err = admin.Exec(ctx, "CREATE SCHEMA "+quoted)
|
||||
requireNoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
_, err := admin.Exec(ctx, "DROP SCHEMA "+quoted+" CASCADE")
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
admin.Close()
|
||||
})
|
||||
config, err := pgxpool.ParseConfig(dsn)
|
||||
requireNoError(t, err)
|
||||
config.ConnConfig.RuntimeParams["search_path"] = schema
|
||||
config.ConnConfig.RuntimeParams["timezone"] = "UTC"
|
||||
pool, err := pgxpool.NewWithConfig(ctx, config)
|
||||
requireNoError(t, err)
|
||||
t.Cleanup(pool.Close)
|
||||
files, err := filepath.Glob("../../../migrations/*.sql")
|
||||
requireNoError(t, err)
|
||||
for _, name := range files {
|
||||
raw, err := os.ReadFile(name)
|
||||
requireNoError(t, err)
|
||||
_, err = pool.Exec(ctx, strings.Split(string(raw), "-- +goose Down")[0])
|
||||
if err != nil {
|
||||
t.Fatalf("migration %s: %v", name, err)
|
||||
}
|
||||
}
|
||||
f := exportFixture{
|
||||
pool: pool,
|
||||
exports: NewGameExportsRepo(pool),
|
||||
games: games_repo.NewGamesRepo(pool),
|
||||
teams: teams_repo.NewTeamsRepo(pool),
|
||||
scenarios: scenarios_repo.NewScenariosRepo(pool),
|
||||
users: users_repo.NewUserRepo(pool),
|
||||
}
|
||||
f.userID, err = f.users.AddUserWithAgreements(ctx, "organizer", "export@test.invalid", "hash", []string{"organizer"},
|
||||
[]user_agreements_repo.Agreement{{AgreementType: "terms", Version: "test"}})
|
||||
requireNoError(t, err)
|
||||
f.scenarioID, err = f.scenarios.AddScenario(ctx, "Original story", f.userID)
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, f.scenarios.UpdateStoryByScenarioID(ctx, f.scenarioID,
|
||||
`{"introduction":{"text":"Intro"},"questions":[{"code":"q","text":"Who?","answer":"Butler"}]}`))
|
||||
f.gameID, err = f.games.AddGame(ctx, "Test export", "Original description", nil, time.Now().Add(12*time.Hour), int32(f.scenarioID), "offline")
|
||||
requireNoError(t, err)
|
||||
f.teamID, err = f.teams.AddTeamWithParticipation(ctx, f.userID, f.gameID, "Team", "original-qr", "captain@test.invalid", 3, false)
|
||||
requireNoError(t, err)
|
||||
return f
|
||||
}
|
||||
|
||||
func requireNoError(t *testing.T, err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (f exportFixture) start(t *testing.T, id string) *Attempt {
|
||||
t.Helper()
|
||||
attempt, err := f.exports.Start(context.Background(), f.gameID, f.userID, id, "https://game.example")
|
||||
requireNoError(t, err)
|
||||
return attempt
|
||||
}
|
||||
@@ -0,0 +1,366 @@
|
||||
// Package game_exports_repo persists export attempts and their immutable snapshots.
|
||||
package game_exports_repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/repos"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrGameNotFound = errors.New("игра не найдена")
|
||||
ErrNotExportable = errors.New("экспорт доступен только для игры в черновике без дат проведения и прогресса")
|
||||
ErrExportNotFound = errors.New("экспорт игры ещё не запущен")
|
||||
ErrOwnershipLost = errors.New("владение экспортом потеряно")
|
||||
)
|
||||
|
||||
const Lease = 2 * time.Minute
|
||||
const PrepareTimeout = 30 * time.Minute
|
||||
|
||||
// Attempt includes private worker data; only GameExport may be exposed to clients.
|
||||
type Attempt struct {
|
||||
repos.GameExport
|
||||
GameID int
|
||||
Snapshot []byte
|
||||
SourceServer string
|
||||
ObjectKey string
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
type GameExportsRepo struct {
|
||||
pool *pgxpool.Pool
|
||||
lease time.Duration
|
||||
prepareTimeout time.Duration
|
||||
}
|
||||
|
||||
func NewGameExportsRepo(pool *pgxpool.Pool) *GameExportsRepo {
|
||||
return &GameExportsRepo{
|
||||
pool: pool,
|
||||
lease: Lease,
|
||||
prepareTimeout: PrepareTimeout,
|
||||
}
|
||||
}
|
||||
|
||||
// Start locks the game before inspecting progress, then saves all package data in one SQL snapshot.
|
||||
func (r *GameExportsRepo) Start(ctx context.Context, gameID, userID int, id, source string) (*Attempt, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
var status string
|
||||
var started, ended *time.Time
|
||||
err = tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
status,
|
||||
started_at,
|
||||
ended_at
|
||||
FROM games
|
||||
WHERE id = $1
|
||||
FOR UPDATE`,
|
||||
gameID,
|
||||
).Scan(&status, &started, &ended)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrGameNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
existing, err := readAttempt(ctx, tx, gameID, true, true)
|
||||
if err == nil {
|
||||
return existing, tx.Commit(ctx)
|
||||
}
|
||||
if !errors.Is(err, ErrExportNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
if status != "draft" || started != nil || ended != nil {
|
||||
return nil, ErrNotExportable
|
||||
}
|
||||
var progress bool
|
||||
err = tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM teams t
|
||||
WHERE t.game_id = $1 AND (
|
||||
EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM actions a
|
||||
WHERE a.team_id = t.id
|
||||
)
|
||||
OR EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM applications a
|
||||
WHERE a.team_id = t.id
|
||||
)
|
||||
OR EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM team_answers a
|
||||
WHERE a.team_id = t.id
|
||||
)
|
||||
)
|
||||
)`,
|
||||
gameID,
|
||||
).Scan(&progress)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if progress {
|
||||
return nil, ErrNotExportable
|
||||
}
|
||||
snapshot, err := captureSnapshot(ctx, tx, gameID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, err = tx.Exec(
|
||||
ctx,
|
||||
`INSERT INTO game_exports (id, game_id, initiator_id, snapshot, source_server, object_key)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)`,
|
||||
id,
|
||||
gameID,
|
||||
userID,
|
||||
snapshot,
|
||||
source,
|
||||
id+".zip",
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := readAttempt(ctx, tx, gameID, true, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return result, tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func (r *GameExportsRepo) Get(ctx context.Context, gameID int) (*Attempt, error) {
|
||||
return readAttempt(ctx, r.pool, gameID, false, false)
|
||||
}
|
||||
|
||||
// IRowReader is implemented by both the pool and a transaction.
|
||||
type IRowReader interface {
|
||||
QueryRow(context.Context, string, ...any) pgx.Row
|
||||
}
|
||||
|
||||
func readAttempt(ctx context.Context, reader IRowReader, gameID int, active, withSnapshot bool) (*Attempt, error) {
|
||||
attempt := &Attempt{}
|
||||
err := reader.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
id,
|
||||
COALESCE(game_id, 0),
|
||||
state,
|
||||
package_id,
|
||||
size,
|
||||
sha256,
|
||||
failure,
|
||||
CASE WHEN $3 THEN snapshot ELSE NULL END,
|
||||
source_server,
|
||||
object_key,
|
||||
created_at
|
||||
FROM game_exports
|
||||
WHERE game_id = $1 AND (NOT $2 OR state IN ('preparing', 'ready'))
|
||||
ORDER BY created_at DESC, id
|
||||
LIMIT 1`,
|
||||
gameID,
|
||||
active,
|
||||
withSnapshot,
|
||||
).Scan(&attempt.ID, &attempt.GameID, &attempt.State, &attempt.PackageID, &attempt.Size,
|
||||
&attempt.SHA256, &attempt.Failure, &attempt.Snapshot, &attempt.SourceServer, &attempt.ObjectKey, &attempt.CreatedAt)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrExportNotFound
|
||||
}
|
||||
return attempt, err
|
||||
}
|
||||
|
||||
// Claim only assigns unowned live attempts. Expired ownership is failed, never reassigned.
|
||||
func (r *GameExportsRepo) Claim(ctx context.Context, owner string) (*Attempt, error) {
|
||||
attempt := &Attempt{}
|
||||
err := r.pool.QueryRow(
|
||||
ctx,
|
||||
`WITH candidate AS (
|
||||
SELECT
|
||||
id
|
||||
FROM game_exports
|
||||
WHERE state = 'preparing' AND owner_id IS NULL
|
||||
AND created_at + $3::interval > clock_timestamp()
|
||||
ORDER BY created_at
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
UPDATE game_exports e
|
||||
SET owner_id = $1, lease_until = clock_timestamp() + $2::interval
|
||||
FROM candidate c
|
||||
WHERE e.id = c.id
|
||||
RETURNING
|
||||
e.id,
|
||||
COALESCE(e.game_id, 0),
|
||||
e.state,
|
||||
e.package_id,
|
||||
e.size,
|
||||
e.sha256,
|
||||
e.failure,
|
||||
e.snapshot,
|
||||
e.source_server,
|
||||
e.object_key,
|
||||
e.created_at`,
|
||||
owner,
|
||||
r.lease.String(),
|
||||
r.prepareTimeout.String(),
|
||||
).Scan(&attempt.ID, &attempt.GameID, &attempt.State, &attempt.PackageID, &attempt.Size,
|
||||
&attempt.SHA256, &attempt.Failure, &attempt.Snapshot, &attempt.SourceServer, &attempt.ObjectKey, &attempt.CreatedAt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return attempt, nil
|
||||
}
|
||||
|
||||
func (r *GameExportsRepo) Heartbeat(ctx context.Context, id, owner string) error {
|
||||
tag, err := r.pool.Exec(
|
||||
ctx,
|
||||
`UPDATE game_exports
|
||||
SET lease_until = clock_timestamp() + $3::interval
|
||||
WHERE id = $1 AND owner_id = $2 AND state = 'preparing'
|
||||
AND lease_until > clock_timestamp()
|
||||
AND created_at + $4::interval > clock_timestamp()`,
|
||||
id,
|
||||
owner,
|
||||
r.lease.String(),
|
||||
r.prepareTimeout.String(),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrOwnershipLost
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Complete fences publication by attempt, owner, lease and total preparation deadline.
|
||||
func (r *GameExportsRepo) Complete(ctx context.Context, id, owner, packageID, digest string, size int64) error {
|
||||
tag, err := r.pool.Exec(
|
||||
ctx,
|
||||
`UPDATE game_exports
|
||||
SET state = 'ready', package_id = $3, sha256 = $4, size = $5,
|
||||
owner_id = NULL, lease_until = NULL
|
||||
WHERE id = $1 AND owner_id = $2 AND state = 'preparing'
|
||||
AND lease_until > clock_timestamp()
|
||||
AND created_at + $6::interval > clock_timestamp()`,
|
||||
id,
|
||||
owner,
|
||||
packageID,
|
||||
digest,
|
||||
size,
|
||||
r.prepareTimeout.String(),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrOwnershipLost
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Fail refuses stale workers without changing game data.
|
||||
func (r *GameExportsRepo) Fail(ctx context.Context, id, owner, reason string) error {
|
||||
tag, err := r.pool.Exec(
|
||||
ctx,
|
||||
`UPDATE game_exports
|
||||
SET state = 'failed', failure = $3, owner_id = NULL, lease_until = NULL
|
||||
WHERE id = $1 AND owner_id = $2
|
||||
AND state = 'preparing' AND lease_until > clock_timestamp()`,
|
||||
id,
|
||||
owner,
|
||||
reason,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrOwnershipLost
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *GameExportsRepo) Expire(ctx context.Context) error {
|
||||
_, err := r.pool.Exec(
|
||||
ctx,
|
||||
`UPDATE game_exports
|
||||
SET state = 'failed', failure = 'Подготовка прервана. Запустите экспорт повторно.',
|
||||
owner_id = NULL, lease_until = NULL
|
||||
WHERE state = 'preparing' AND (
|
||||
lease_until <= clock_timestamp() OR created_at + $1::interval <= clock_timestamp()
|
||||
)`,
|
||||
r.prepareTimeout.String(),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// CleanupKeys retries deletions until all old workers' preparation deadlines have elapsed.
|
||||
func (r *GameExportsRepo) CleanupKeys(ctx context.Context) ([]string, error) {
|
||||
rows, err := r.pool.Query(
|
||||
ctx,
|
||||
`SELECT
|
||||
object_key
|
||||
FROM game_exports
|
||||
WHERE state = 'failed' AND (cleaned_at IS NULL OR (
|
||||
cleaned_at < created_at + $1::interval AND cleaned_at < clock_timestamp() - INTERVAL '2 minutes'
|
||||
))
|
||||
ORDER BY created_at
|
||||
LIMIT 100`,
|
||||
(r.prepareTimeout + r.lease).String(),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
keys := []string{}
|
||||
for rows.Next() {
|
||||
var key string
|
||||
if err := rows.Scan(&key); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keys = append(keys, key)
|
||||
}
|
||||
return keys, rows.Err()
|
||||
}
|
||||
|
||||
func (r *GameExportsRepo) MarkCleaned(ctx context.Context, key string) error {
|
||||
_, err := r.pool.Exec(
|
||||
ctx,
|
||||
`UPDATE game_exports
|
||||
SET cleaned_at = clock_timestamp()
|
||||
WHERE object_key = $1 AND state = 'failed'`,
|
||||
key,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// IsInactive only identifies known terminal attempts, including leftovers after successful publication.
|
||||
func (r *GameExportsRepo) IsInactive(ctx context.Context, id string) (bool, error) {
|
||||
var inactive bool
|
||||
err := r.pool.QueryRow(
|
||||
ctx,
|
||||
`SELECT
|
||||
EXISTS (
|
||||
SELECT
|
||||
1
|
||||
FROM game_exports
|
||||
WHERE id = $1 AND state IN ('ready', 'failed')
|
||||
)`,
|
||||
id,
|
||||
).Scan(&inactive)
|
||||
return inactive, err
|
||||
}
|
||||
@@ -0,0 +1,340 @@
|
||||
package game_exports_repo
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/repos"
|
||||
"evening_detective_server/internal/repos/actions_repo"
|
||||
"evening_detective_server/internal/repos/applications_repo"
|
||||
"evening_detective_server/internal/repos/game_map_repo"
|
||||
"evening_detective_server/internal/repos/game_reviews_repo"
|
||||
"evening_detective_server/internal/repos/game_screen_plans_repo"
|
||||
"evening_detective_server/internal/repos/playlists_repo"
|
||||
"evening_detective_server/internal/repos/screen_sessions_repo"
|
||||
"evening_detective_server/internal/repos/team_answers_repo"
|
||||
"evening_detective_server/internal/repos/tickets_repo"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
func TestExportSnapshotAndEditableGame(t *testing.T) {
|
||||
for _, state := range []string{"preparing", "ready"} {
|
||||
t.Run(state, func(t *testing.T) {
|
||||
f := newExportFixture(t)
|
||||
ctx := context.Background()
|
||||
music := playlists_repo.New(f.pool)
|
||||
base, err := music.Create(ctx, "Scenario default")
|
||||
requireNoError(t, err)
|
||||
assigned, err := music.Create(ctx, "Game assigned")
|
||||
requireNoError(t, err)
|
||||
selected, err := music.Create(ctx, "Selected by screen")
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, music.SetScenario(ctx, f.scenarioID, base.ID))
|
||||
requireNoError(t, music.SetGame(ctx, f.gameID, assigned.ID))
|
||||
_, err = music.SetMusic(ctx, f.gameID, selected.ID, 0, 0, 100, false, 1)
|
||||
requireNoError(t, err)
|
||||
ticketRepo := tickets_repo.New(f.pool)
|
||||
activeHash := sha256.Sum256([]byte("active-ticket"))
|
||||
usedHash := sha256.Sum256([]byte("used-ticket"))
|
||||
_, err = ticketRepo.CreateBatch(ctx, f.gameID, [][]byte{activeHash[:], usedHash[:]})
|
||||
requireNoError(t, err)
|
||||
_, err = ticketRepo.Accept(ctx, usedHash[:], f.userID)
|
||||
requireNoError(t, err)
|
||||
token := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
requireNoError(t, f.teams.SetParticipationToken(ctx, f.teamID, token))
|
||||
mapRepo := game_map_repo.NewGameMapRepo(f.pool)
|
||||
requireNoError(t, mapRepo.Ensure(ctx, f.gameID, "team", f.teamID))
|
||||
plans := game_screen_plans_repo.NewGameScreenPlansRepo(f.pool)
|
||||
_, err = plans.Save(ctx, f.gameID, []game_screen_plans_repo.Step{{Mode: 1}}, 0)
|
||||
requireNoError(t, err)
|
||||
screens := screen_sessions_repo.NewScreenSessionsRepo(f.pool)
|
||||
screenID, err := screens.Create(ctx, "screen-token-hash")
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, screens.Bind(ctx, screenID, f.gameID, "screen-token-hash"))
|
||||
reviews := game_reviews_repo.NewGameReviewsRepo(f.pool)
|
||||
review, err := reviews.Create(ctx, f.gameID, f.teamID, 5, 2, "Review")
|
||||
requireNoError(t, err)
|
||||
attempt := f.start(t, "attempt-original")
|
||||
if state == "ready" {
|
||||
_, err := f.exports.Claim(ctx, "worker")
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, f.exports.Complete(ctx, attempt.ID, "worker", "package-id", "digest", 123))
|
||||
}
|
||||
var snapshot struct {
|
||||
Game struct {
|
||||
PlaylistID int `json:"playlist_id"`
|
||||
}
|
||||
Scenario struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
Teams []struct {
|
||||
Password string `json:"password"`
|
||||
}
|
||||
Tickets []struct {
|
||||
TokenHash []byte `json:"token_hash"`
|
||||
Status string `json:"status"`
|
||||
}
|
||||
Playlists []struct {
|
||||
ID int `json:"id"`
|
||||
}
|
||||
}
|
||||
requireNoError(t, json.Unmarshal(attempt.Snapshot, &snapshot))
|
||||
if snapshot.Game.PlaylistID != selected.ID || len(snapshot.Playlists) != 2 || len(snapshot.Tickets) != 2 ||
|
||||
!bytes.Equal(snapshot.Tickets[0].TokenHash, activeHash[:]) || snapshot.Teams[0].Password != "original-qr" {
|
||||
t.Fatalf("snapshot lost effective music, ticket or QR: %+v", snapshot)
|
||||
}
|
||||
requireNoError(t, f.teams.UpdateTeam(ctx, f.teamID, "Online edit"))
|
||||
_, err = f.teams.RegisterTeam(ctx, f.gameID, "delivery", "New", "new@test.invalid", 2, "password")
|
||||
requireNoError(t, err)
|
||||
_, err = f.teams.AnswerParticipation(ctx, token, "confirmed")
|
||||
requireNoError(t, err)
|
||||
_, err = actions_repo.NewActionsRepo(f.pool).AddAction(ctx, f.teamID, "code", false)
|
||||
requireNoError(t, err)
|
||||
_, err = ticketRepo.Accept(ctx, activeHash[:], f.userID)
|
||||
requireNoError(t, err)
|
||||
_, err = music.SetMusic(ctx, f.gameID, base.ID, 0, 0, 100, false, 1)
|
||||
requireNoError(t, err)
|
||||
_, err = plans.Save(ctx, f.gameID, []game_screen_plans_repo.Step{{Mode: 2}}, 0)
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, screens.SetMode(ctx, f.gameID, 2))
|
||||
requireNoError(t, screens.Unbind(ctx, screenID))
|
||||
requireNoError(t, mapRepo.Update(ctx, game_map_repo.Item{GameID: f.gameID, Kind: "team", ItemID: f.teamID, X: 10}))
|
||||
_, err = reviews.Moderate(ctx, review.ID, "approved")
|
||||
requireNoError(t, err)
|
||||
same, err := f.exports.Start(ctx, f.gameID, f.userID, "not-a-new-attempt", "test")
|
||||
requireNoError(t, err)
|
||||
if same.ID != attempt.ID || !bytes.Equal(same.Snapshot, attempt.Snapshot) {
|
||||
t.Fatal("online edits changed an existing export snapshot")
|
||||
}
|
||||
requireNoError(t, f.scenarios.UpdateScenarioByID(ctx, f.scenarioID, "Edited shared story", "", ""))
|
||||
_, err = music.Update(ctx, selected.ID, "Edited shared playlist")
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, f.users.DeleteUser(ctx, f.userID))
|
||||
team, err := f.teams.GetTeamByID(ctx, f.teamID)
|
||||
requireNoError(t, err)
|
||||
if team.Password != "original-qr" {
|
||||
t.Fatal("account deletion changed exported team")
|
||||
}
|
||||
ticket, err := ticketRepo.GetByHash(ctx, usedHash[:])
|
||||
requireNoError(t, err)
|
||||
if ticket.UsedBy != nil || ticket.Status != "used" {
|
||||
t.Fatal("account deletion changed exported ticket")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExportConcurrentStartAndTeamWrites(t *testing.T) {
|
||||
for _, tc := range []struct{ name string }{{"one active attempt and consistent team snapshot"}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newExportFixture(t)
|
||||
ctx := context.Background()
|
||||
var wg sync.WaitGroup
|
||||
attempts := make(chan *Attempt, 8)
|
||||
errs := make(chan error, 9)
|
||||
for n := range 8 {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
attempt, err := f.exports.Start(ctx, f.gameID, f.userID, fmt.Sprintf("attempt-%d", n), "https://game.example")
|
||||
if err != nil {
|
||||
errs <- err
|
||||
} else {
|
||||
attempts <- attempt
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Add(1)
|
||||
var changeErr error
|
||||
go func() { defer wg.Done(); changeErr = f.teams.UpdateTeam(ctx, f.teamID, "Concurrent name") }()
|
||||
wg.Wait()
|
||||
close(errs)
|
||||
close(attempts)
|
||||
for err := range errs {
|
||||
t.Error(err)
|
||||
}
|
||||
if changeErr != nil {
|
||||
t.Fatal(changeErr)
|
||||
}
|
||||
var id string
|
||||
for attempt := range attempts {
|
||||
if id != "" && id != attempt.ID {
|
||||
t.Fatal("more than one active export")
|
||||
}
|
||||
id = attempt.ID
|
||||
var snapshot struct {
|
||||
Teams []struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
}
|
||||
requireNoError(t, json.Unmarshal(attempt.Snapshot, &snapshot))
|
||||
if snapshot.Teams[0].Name != "Concurrent name" && snapshot.Teams[0].Name != "Team" {
|
||||
t.Fatalf("unexpected team snapshot: %s", snapshot.Teams[0].Name)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExportRejectsProgressAndRunningGames(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
mutate func(exportFixture) error
|
||||
}{
|
||||
{"started", func(f exportFixture) error { return f.games.StartGame(context.Background(), f.gameID) }},
|
||||
{"actions", func(f exportFixture) error {
|
||||
_, e := actions_repo.NewActionsRepo(f.pool).AddAction(context.Background(), f.teamID, "p", false)
|
||||
return e
|
||||
}},
|
||||
{"applications", func(f exportFixture) error {
|
||||
_, e := applications_repo.NewApplicationsRepo(f.pool).AddApplication(context.Background(), f.teamID, "Clue", "clue.pdf", "pdf")
|
||||
return e
|
||||
}},
|
||||
{"answers", func(f exportFixture) error {
|
||||
return team_answers_repo.NewTeamAnswersRepo(f.pool).UpsertAnswers(context.Background(), f.teamID, []repos.TeamAnswer{{QuestionCode: "q"}})
|
||||
}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newExportFixture(t)
|
||||
requireNoError(t, tc.mutate(f))
|
||||
_, err := f.exports.Start(context.Background(), f.gameID, f.userID, "attempt", "https://game.example")
|
||||
if !errors.Is(err, ErrNotExportable) {
|
||||
t.Fatalf("export allowed: %v", err)
|
||||
}
|
||||
requireNoError(t, f.teams.UpdateTeam(context.Background(), f.teamID, "Still writable"))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpiredExportCannotPublishAndRetryTakesNewSnapshot(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
lease, timeout time.Duration
|
||||
}{
|
||||
{"lease", 100 * time.Millisecond, time.Minute},
|
||||
{"total deadline", time.Minute, 100 * time.Millisecond},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newExportFixture(t)
|
||||
f.exports.lease, f.exports.prepareTimeout = tc.lease, tc.timeout
|
||||
ctx := context.Background()
|
||||
attempt := f.start(t, "old-attempt")
|
||||
claimed, err := f.exports.Claim(ctx, "old-worker")
|
||||
requireNoError(t, err)
|
||||
if claimed.ID != attempt.ID {
|
||||
t.Fatal("wrong claimed attempt")
|
||||
}
|
||||
_, err = NewGameExportsRepo(f.pool).Claim(ctx, "other-worker")
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
t.Fatalf("ownership stolen: %v", err)
|
||||
}
|
||||
if err := f.exports.Complete(ctx, attempt.ID, "other-worker", "id", "digest", 10); !errors.Is(err, ErrOwnershipLost) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(150 * time.Millisecond)
|
||||
if err := f.exports.Heartbeat(ctx, attempt.ID, "old-worker"); !errors.Is(err, ErrOwnershipLost) {
|
||||
t.Fatalf("expired heartbeat accepted: %v", err)
|
||||
}
|
||||
requireNoError(t, f.exports.Expire(ctx))
|
||||
failed, err := f.exports.Get(ctx, f.gameID)
|
||||
requireNoError(t, err)
|
||||
if failed.State != "failed" {
|
||||
t.Fatalf("not recovered: %+v", failed)
|
||||
}
|
||||
requireNoError(t, f.teams.UpdateTeam(ctx, f.teamID, "New snapshot name"))
|
||||
f.exports.lease, f.exports.prepareTimeout = Lease, PrepareTimeout
|
||||
retry := f.start(t, "new-attempt")
|
||||
if retry.ID == attempt.ID || bytes.Equal(retry.Snapshot, attempt.Snapshot) {
|
||||
t.Fatal("retry reused old snapshot")
|
||||
}
|
||||
if err := f.exports.Fail(ctx, attempt.ID, "old-worker", "stale failure"); !errors.Is(err, ErrOwnershipLost) {
|
||||
t.Fatalf("stale failure accepted: %v", err)
|
||||
}
|
||||
current, err := f.exports.Get(ctx, f.gameID)
|
||||
requireNoError(t, err)
|
||||
if current.ID != retry.ID || current.State != "preparing" {
|
||||
t.Fatal("stale failure changed retry")
|
||||
}
|
||||
|
||||
if err := f.exports.Complete(ctx, attempt.ID, "old-worker", "id", "digest", 10); !errors.Is(err, ErrOwnershipLost) {
|
||||
t.Fatalf("stale publication accepted: %v", err)
|
||||
}
|
||||
_, err = NewGameExportsRepo(f.pool).Claim(ctx, "new-worker")
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, f.exports.Heartbeat(ctx, retry.ID, "new-worker"))
|
||||
requireNoError(t, f.exports.Complete(ctx, retry.ID, "new-worker", "package-id", "digest", 10))
|
||||
ready, err := NewGameExportsRepo(f.pool).Get(ctx, f.gameID)
|
||||
requireNoError(t, err)
|
||||
if ready.State != "ready" {
|
||||
t.Fatal("ready package lost after repository restart")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFailedExportCleanupSurvivesGameDeletion(t *testing.T) {
|
||||
for _, tc := range []struct{ name string }{{"cleanup remains durable after game removal"}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newExportFixture(t)
|
||||
ctx := context.Background()
|
||||
attempt := f.start(t, "failed-attempt")
|
||||
_, err := f.exports.Claim(ctx, "worker")
|
||||
requireNoError(t, err)
|
||||
requireNoError(t, f.exports.Fail(ctx, attempt.ID, "worker", "Safe failure"))
|
||||
requireNoError(t, f.teams.DeleteTeam(ctx, f.teamID))
|
||||
requireNoError(t, f.games.DeleteGameByID(ctx, f.gameID))
|
||||
keys, err := f.exports.CleanupKeys(ctx)
|
||||
requireNoError(t, err)
|
||||
if len(keys) != 1 || keys[0] != attempt.ObjectKey {
|
||||
t.Fatal("deleted game lost pending object cleanup")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreparingExportSurvivesGameDeletion(t *testing.T) {
|
||||
for _, result := range []string{"ready", "failed", "expired"} {
|
||||
t.Run(result, func(t *testing.T) {
|
||||
f := newExportFixture(t)
|
||||
ctx := context.Background()
|
||||
attempt := f.start(t, "deleted-game-attempt")
|
||||
requireNoError(t, f.teams.DeleteTeam(ctx, f.teamID))
|
||||
requireNoError(t, f.games.DeleteGameByID(ctx, f.gameID))
|
||||
claimed, err := f.exports.Claim(ctx, "worker")
|
||||
requireNoError(t, err)
|
||||
if claimed.ID != attempt.ID || claimed.GameID != 0 || !bytes.Equal(claimed.Snapshot, attempt.Snapshot) {
|
||||
t.Fatal("deleting the game damaged queued snapshot")
|
||||
}
|
||||
switch result {
|
||||
case "ready":
|
||||
requireNoError(t, f.exports.Complete(ctx, attempt.ID, "worker", "package-id", "digest", 123))
|
||||
case "failed":
|
||||
requireNoError(t, f.exports.Fail(ctx, attempt.ID, "worker", "Safe failure"))
|
||||
case "expired":
|
||||
f.exports.prepareTimeout = 0
|
||||
requireNoError(t, f.exports.Expire(ctx))
|
||||
}
|
||||
inactive, err := f.exports.IsInactive(ctx, attempt.ID)
|
||||
requireNoError(t, err)
|
||||
if !inactive {
|
||||
t.Fatal("deleted game left an unfinishable export")
|
||||
}
|
||||
keys, err := f.exports.CleanupKeys(ctx)
|
||||
requireNoError(t, err)
|
||||
expected := 1
|
||||
if result == "ready" {
|
||||
expected = 0
|
||||
}
|
||||
if len(keys) != expected {
|
||||
t.Fatalf("wrong object cleanup state: %v", keys)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package game_exports_repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// captureSnapshot uses one statement so editable shared scenarios and playlists are consistent.
|
||||
func captureSnapshot(ctx context.Context, tx pgx.Tx, gameID int) ([]byte, error) {
|
||||
var raw []byte
|
||||
err := tx.QueryRow(
|
||||
ctx,
|
||||
`SELECT jsonb_build_object(
|
||||
'game', jsonb_build_object(
|
||||
'id', g.id, 'name', g.name, 'description', COALESCE(g.description, ''),
|
||||
'location', g.location, 'start_at', g.start_at AT TIME ZONE 'UTC',
|
||||
'status', g.status, 'format', g.format, 'scenario_id', g.scenario_id,
|
||||
'playlist_id', COALESCE(m.playlist_id, g.playlist_id),
|
||||
'started_at', g.started_at AT TIME ZONE 'UTC', 'ended_at', g.ended_at AT TIME ZONE 'UTC'
|
||||
),
|
||||
'scenario', jsonb_build_object(
|
||||
'id', s.id, 'name', s.name, 'description', s.description,
|
||||
'image', COALESCE(s.image, ''), 'story', s.scenario, 'author_id', s.author_id,
|
||||
'status', s.status, 'created_at', s.created_at AT TIME ZONE 'UTC',
|
||||
'updated_at', s.updated_at AT TIME ZONE 'UTC',
|
||||
'published_at', s.published_at AT TIME ZONE 'UTC', 'playlist_id', s.playlist_id
|
||||
),
|
||||
'teams', COALESCE((
|
||||
SELECT jsonb_agg(jsonb_build_object(
|
||||
'id', t.id, 'game_id', t.game_id, 'name', t.name, 'status', t.status,
|
||||
'password', t.password, 'creator_id', t.creator_id, 'captain_email', t.captain_email,
|
||||
'participants_count', t.participants_count, 'participation_status', t.participation_status,
|
||||
'participation_source', t.participation_source, 'participation_answered_at', t.participation_answered_at,
|
||||
'registration_delivery_id', t.registration_delivery_id, 'created_at', t.created_at AT TIME ZONE 'UTC'
|
||||
) ORDER BY t.id)
|
||||
FROM teams t
|
||||
WHERE t.game_id = g.id
|
||||
), '[]'::jsonb),
|
||||
'tickets', COALESCE((
|
||||
SELECT jsonb_agg(jsonb_build_object(
|
||||
'id', t.id, 'game_id', t.game_id, 'token_hash', encode(t.token_hash, 'base64'),
|
||||
'status', t.status, 'used_at', t.used_at, 'used_by', t.used_by, 'created_at', t.created_at
|
||||
) ORDER BY t.id)
|
||||
FROM tickets t
|
||||
WHERE t.game_id = g.id
|
||||
), '[]'::jsonb),
|
||||
'playlists', COALESCE((
|
||||
SELECT jsonb_agg(jsonb_build_object(
|
||||
'id', p.id, 'name', p.name, 'tracks', COALESCE((
|
||||
SELECT jsonb_agg(jsonb_build_object(
|
||||
'id', tr.id, 'name', tr.name, 'filename', tr.filename, 'position', tr.position
|
||||
) ORDER BY tr.position, tr.id)
|
||||
FROM playlist_tracks tr
|
||||
WHERE tr.playlist_id = p.id
|
||||
), '[]'::jsonb)
|
||||
) ORDER BY p.id)
|
||||
FROM playlists p
|
||||
WHERE p.id IN (COALESCE(m.playlist_id, g.playlist_id), s.playlist_id)
|
||||
), '[]'::jsonb),
|
||||
'screen_plan', (
|
||||
SELECT jsonb_build_object('game_id', sp.game_id, 'steps', sp.steps, 'current_step', sp.current_step)
|
||||
FROM game_screen_plans sp
|
||||
WHERE sp.game_id = g.id
|
||||
),
|
||||
'map', COALESCE((
|
||||
SELECT jsonb_agg(jsonb_build_object(
|
||||
'game_id', mi.game_id, 'kind', mi.kind, 'item_id', mi.item_id,
|
||||
'x', mi.x, 'y', mi.y, 'hidden', mi.hidden
|
||||
) ORDER BY mi.kind, mi.item_id)
|
||||
FROM game_map_items mi
|
||||
WHERE mi.game_id = g.id
|
||||
), '[]'::jsonb)
|
||||
)
|
||||
FROM games g
|
||||
JOIN scenarios s ON s.id = g.scenario_id
|
||||
LEFT JOIN game_music_states m ON m.game_id = g.id
|
||||
WHERE g.id = $1`,
|
||||
gameID,
|
||||
).Scan(&raw)
|
||||
return raw, err
|
||||
}
|
||||
@@ -52,7 +52,7 @@ func (r *Repo) Create(ctx context.Context, gameID, teamID int, rating, difficult
|
||||
FROM teams t
|
||||
WHERE t.id = $2 AND t.game_id = $1
|
||||
ON CONFLICT (game_id, team_id) DO NOTHING
|
||||
RETURNING id, game_id, team_id, rating, text, status, created_at`,
|
||||
RETURNING id, game_id, team_id, rating, difficulty, text, status, created_at`,
|
||||
gameID,
|
||||
teamID,
|
||||
rating,
|
||||
|
||||
@@ -76,7 +76,16 @@ func (r *GamesRepo) GetGameByID(
|
||||
games.status,
|
||||
games.scenario_id,
|
||||
games.started_at,
|
||||
games.ended_at
|
||||
games.ended_at,
|
||||
(SELECT
|
||||
jsonb_build_object(
|
||||
'id', e.id, 'state', e.state, 'package_id', e.package_id,
|
||||
'size', e.size, 'sha256', e.sha256, 'failure', e.failure
|
||||
)
|
||||
FROM game_exports e
|
||||
WHERE e.game_id = games.id
|
||||
ORDER BY e.created_at DESC, e.id
|
||||
LIMIT 1)
|
||||
FROM games
|
||||
WHERE id = $1`,
|
||||
id,
|
||||
@@ -92,6 +101,7 @@ func (r *GamesRepo) GetGameByID(
|
||||
&game.ScenarioID,
|
||||
&game.StartedAt,
|
||||
&game.EndedAt,
|
||||
&game.Export,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
@@ -136,7 +146,16 @@ func (r *GamesRepo) GetGames(
|
||||
games.location,
|
||||
games.start_at,
|
||||
games.format,
|
||||
games.status
|
||||
games.status,
|
||||
(SELECT
|
||||
jsonb_build_object(
|
||||
'id', e.id, 'state', e.state, 'package_id', e.package_id,
|
||||
'size', e.size, 'sha256', e.sha256, 'failure', e.failure
|
||||
)
|
||||
FROM game_exports e
|
||||
WHERE e.game_id = games.id
|
||||
ORDER BY e.created_at DESC, e.id
|
||||
LIMIT 1)
|
||||
FROM games
|
||||
ORDER BY created_at DESC`,
|
||||
)
|
||||
@@ -156,6 +175,7 @@ func (r *GamesRepo) GetGames(
|
||||
&game.StartAt,
|
||||
&game.Format,
|
||||
&game.Status,
|
||||
&game.Export,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -28,6 +28,7 @@ func TestAnswerAccessPostgres(t *testing.T) {
|
||||
}
|
||||
defer pool.Close()
|
||||
_, err = pool.Exec(ctx, `
|
||||
CREATE TEMP TABLE game_exports(id TEXT, game_id INT, state TEXT, package_id TEXT, size BIGINT, sha256 TEXT, failure TEXT, created_at TIMESTAMPTZ);
|
||||
CREATE TEMP TABLE games (id INT, name TEXT, description TEXT, location TEXT, start_at TIMESTAMP,
|
||||
format TEXT, status TEXT, scenario_id INT, started_at TIMESTAMP, ended_at TIMESTAMP);
|
||||
CREATE TEMP TABLE teams (id INT, name TEXT, captain_email TEXT, participants_count INT,
|
||||
@@ -71,10 +72,10 @@ func TestAnswerAccessPostgres(t *testing.T) {
|
||||
if _, err := pool.Exec(ctx, "UPDATE teams SET participation_status=$1", tc.participation); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := service.GetTeamAnswers(ctx, 1, tc.password); !errors.Is(err, teams_repo.ErrTeamNotFound) {
|
||||
if _, err := service.GetTeamAnswers(ctx, 1, tc.password); !errors.Is(err, ErrTeamNotFound) {
|
||||
t.Fatalf("read access: %v", err)
|
||||
}
|
||||
if err := service.SubmitTeamAnswers(ctx, 1, tc.password, nil); !errors.Is(err, teams_repo.ErrTeamNotFound) {
|
||||
if err := service.SubmitTeamAnswers(ctx, 1, tc.password, nil); !errors.Is(err, ErrTeamNotFound) {
|
||||
t.Fatalf("write access: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
package game_export_service
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"evening_detective_server/internal/repos/game_exports_repo"
|
||||
"evening_detective_server/internal/services/game_package_service"
|
||||
)
|
||||
|
||||
func mapSnapshot(attempt *game_exports_repo.Attempt) (*game_package_service.Snapshot, error) {
|
||||
var data game_package_service.Data
|
||||
if err := json.Unmarshal(attempt.Snapshot, &data); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &game_package_service.Snapshot{
|
||||
SourceServer: attempt.SourceServer,
|
||||
Data: data,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,394 @@
|
||||
package game_export_service_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/app"
|
||||
"evening_detective_server/internal/modules/file_storage"
|
||||
"evening_detective_server/internal/modules/processor_jwt"
|
||||
"evening_detective_server/internal/repos/actions_repo"
|
||||
"evening_detective_server/internal/repos/applications_repo"
|
||||
"evening_detective_server/internal/repos/game_exports_repo"
|
||||
"evening_detective_server/internal/repos/game_map_repo"
|
||||
"evening_detective_server/internal/repos/games_repo"
|
||||
"evening_detective_server/internal/repos/scenarios_repo"
|
||||
"evening_detective_server/internal/repos/screen_sessions_repo"
|
||||
"evening_detective_server/internal/repos/teams_repo"
|
||||
"evening_detective_server/internal/repos/tickets_repo"
|
||||
"evening_detective_server/internal/repos/user_agreements_repo"
|
||||
"evening_detective_server/internal/repos/users_repo"
|
||||
"evening_detective_server/internal/services/game_export_service"
|
||||
"evening_detective_server/internal/services/game_package_service"
|
||||
"evening_detective_server/internal/services/game_service"
|
||||
"evening_detective_server/internal/services/map_service"
|
||||
"evening_detective_server/internal/services/screen_service"
|
||||
"evening_detective_server/proto"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type memoryStorage struct {
|
||||
mu sync.Mutex
|
||||
objects map[string][]byte
|
||||
uploadErr error
|
||||
deletes int
|
||||
}
|
||||
|
||||
func (s *memoryStorage) PutStream(ctx context.Context, key string, body io.Reader, mime string) error {
|
||||
data, err := io.ReadAll(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.objects[key] = data
|
||||
return s.uploadErr
|
||||
}
|
||||
|
||||
func (s *memoryStorage) Open(ctx context.Context, key string) (io.ReadCloser, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
data, ok := s.objects[key]
|
||||
if !ok {
|
||||
return nil, file_storage.ErrFileNotFound
|
||||
}
|
||||
return io.NopCloser(bytes.NewReader(data)), nil
|
||||
}
|
||||
|
||||
func (s *memoryStorage) Delete(ctx context.Context, key string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
delete(s.objects, key)
|
||||
s.deletes++
|
||||
return nil
|
||||
}
|
||||
|
||||
type serviceFixture struct {
|
||||
pool *pgxpool.Pool
|
||||
repo *game_exports_repo.GameExportsRepo
|
||||
codec *game_package_service.GamePackageService
|
||||
games *games_repo.GamesRepo
|
||||
teams *teams_repo.TeamsRepo
|
||||
scenarios *scenarios_repo.ScenariosRepo
|
||||
gameID, teamID, scenarioID, userID int
|
||||
media []byte
|
||||
}
|
||||
|
||||
// newServiceFixture is schema infrastructure only; all fixture data uses public repositories.
|
||||
func newServiceFixture(t *testing.T) serviceFixture {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("set TEST_DATABASE_URL for export lifecycle checks")
|
||||
}
|
||||
ctx := context.Background()
|
||||
admin, err := pgxpool.New(ctx, dsn)
|
||||
check(t, err)
|
||||
schema := fmt.Sprintf("export_service_test_%d", time.Now().UnixNano())
|
||||
quoted := pgx.Identifier{schema}.Sanitize()
|
||||
_, err = admin.Exec(ctx, "CREATE SCHEMA "+quoted)
|
||||
check(t, err)
|
||||
t.Cleanup(func() {
|
||||
_, err := admin.Exec(ctx, "DROP SCHEMA "+quoted+" CASCADE")
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
admin.Close()
|
||||
})
|
||||
config, err := pgxpool.ParseConfig(dsn)
|
||||
check(t, err)
|
||||
config.ConnConfig.RuntimeParams["search_path"] = schema
|
||||
config.ConnConfig.RuntimeParams["timezone"] = "UTC"
|
||||
pool, err := pgxpool.NewWithConfig(ctx, config)
|
||||
check(t, err)
|
||||
t.Cleanup(pool.Close)
|
||||
migrations, err := filepath.Glob("../../../migrations/*.sql")
|
||||
check(t, err)
|
||||
for _, name := range migrations {
|
||||
raw, err := os.ReadFile(name)
|
||||
check(t, err)
|
||||
_, err = pool.Exec(ctx, strings.Split(string(raw), "-- +goose Down")[0])
|
||||
if err != nil {
|
||||
t.Fatalf("migration %s: %v", name, err)
|
||||
}
|
||||
}
|
||||
f := serviceFixture{
|
||||
pool: pool,
|
||||
repo: game_exports_repo.NewGameExportsRepo(pool),
|
||||
codec: game_package_service.NewGamePackageService("https://files.example/api/files/"),
|
||||
games: games_repo.NewGamesRepo(pool),
|
||||
teams: teams_repo.NewTeamsRepo(pool),
|
||||
scenarios: scenarios_repo.NewScenariosRepo(pool),
|
||||
media: []byte("\x89PNG\r\n\x1a\noriginal-image"),
|
||||
}
|
||||
f.userID, err = users_repo.NewUserRepo(pool).AddUserWithAgreements(ctx, "organizer", "export@test.invalid", "hash", []string{"organizer"},
|
||||
[]user_agreements_repo.Agreement{{AgreementType: "terms", Version: "test"}})
|
||||
check(t, err)
|
||||
f.scenarioID, err = f.scenarios.AddScenario(ctx, "Saved story", f.userID)
|
||||
check(t, err)
|
||||
check(t, f.scenarios.UpdateStoryByScenarioID(ctx, f.scenarioID,
|
||||
`{"background":"original.png","introduction":{"text":"Saved intro"},"questions":[{"code":"q","text":"Who?","answer":"Butler"}]}`))
|
||||
f.gameID, err = f.games.AddGame(ctx, "Test game", "Description", nil, time.Now().Add(48*time.Hour), int32(f.scenarioID), "offline")
|
||||
check(t, err)
|
||||
f.teamID, err = f.teams.AddTeam(ctx, f.userID, f.gameID, "Original team", "original-qr")
|
||||
check(t, err)
|
||||
hash := sha256.Sum256([]byte("ticket"))
|
||||
_, err = tickets_repo.New(pool).CreateBatch(ctx, f.gameID, [][]byte{hash[:]})
|
||||
check(t, err)
|
||||
return f
|
||||
}
|
||||
|
||||
func check(t *testing.T, err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (f serviceFixture) service(storage file_storage.IStreamStorage) *game_export_service.GameExportService {
|
||||
return game_export_service.NewGameExportService(f.repo, f.codec, storage, func(ctx context.Context, key string) (io.ReadCloser, error) {
|
||||
if key != "original.png" {
|
||||
return nil, file_storage.ErrFileNotFound
|
||||
}
|
||||
return io.NopCloser(bytes.NewReader(f.media)), nil
|
||||
}, "https://game.example", false)
|
||||
}
|
||||
|
||||
func download(t *testing.T, service *game_export_service.GameExportService, gameID int) (*game_export_service.GameExport, []byte) {
|
||||
t.Helper()
|
||||
info, body, err := service.OpenDownload(context.Background(), gameID)
|
||||
check(t, err)
|
||||
defer func() { _ = body.Close() }()
|
||||
data, err := io.ReadAll(body)
|
||||
check(t, err)
|
||||
return info, data
|
||||
}
|
||||
|
||||
func verifyPackage(t *testing.T, f serviceFixture, service *game_export_service.GameExportService) []byte {
|
||||
t.Helper()
|
||||
info, data := download(t, service, f.gameID)
|
||||
digest := sha256.Sum256(data)
|
||||
if info.Size != int64(len(data)) || info.SHA256 != hex.EncodeToString(digest[:]) {
|
||||
t.Fatal("incorrect archive size or digest")
|
||||
}
|
||||
bundle, err := f.codec.Unpack(context.Background(), bytes.NewReader(data), int64(len(data)), t.TempDir())
|
||||
check(t, err)
|
||||
defer func() { _ = os.RemoveAll(bundle.Dir) }()
|
||||
if bundle.Manifest.PackageID != info.PackageID || bundle.Data.Teams[0].Password != "original-qr" ||
|
||||
len(bundle.Data.Tickets) != 1 || bundle.Data.Scenario.Story.Introduction.Text != "Saved intro" {
|
||||
t.Fatal("package lost saved snapshot or actual manifest ID")
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
func TestExportBuildsSavedSnapshotAcrossWorkerRestart(t *testing.T) {
|
||||
for _, tc := range []struct{ name string }{{"snapshot, durable bytes and abort"}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newServiceFixture(t)
|
||||
storage := &memoryStorage{objects: map[string][]byte{}}
|
||||
firstWorker := f.service(storage)
|
||||
original, err := firstWorker.Start(context.Background(), f.gameID, f.userID)
|
||||
check(t, err)
|
||||
_, _, err = firstWorker.OpenDownload(context.Background(), f.gameID)
|
||||
if !errors.Is(err, game_export_service.ErrNotReady) {
|
||||
t.Fatal("preparing archive downloadable")
|
||||
}
|
||||
check(t, f.scenarios.UpdateStoryByScenarioID(context.Background(), f.scenarioID,
|
||||
`{"introduction":{"text":"Changed after snapshot"}}`))
|
||||
restartedWorker := f.service(storage)
|
||||
worked, err := restartedWorker.ProcessNext(context.Background())
|
||||
check(t, err)
|
||||
if !worked {
|
||||
t.Fatal("pending export lost on restart")
|
||||
}
|
||||
first := verifyPackage(t, f, restartedWorker)
|
||||
verifyExportedGameReads(t, f, restartedWorker, original.ID)
|
||||
repeated, err := restartedWorker.Start(context.Background(), f.gameID, f.userID)
|
||||
check(t, err)
|
||||
if repeated.ID != original.ID || repeated.State != "ready" {
|
||||
t.Fatal("ready export not idempotent")
|
||||
}
|
||||
_, second := download(t, f.service(storage), f.gameID)
|
||||
if !bytes.Equal(first, second) {
|
||||
t.Fatal("download changed after restart")
|
||||
}
|
||||
assertAbortedDownload(t, f, restartedWorker, first)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func assertAbortedDownload(t *testing.T, f serviceFixture, service *game_export_service.GameExportService, original []byte) {
|
||||
t.Helper()
|
||||
jwt := processor_jwt.NewProcessor("export-test-secret")
|
||||
token, err := jwt.GenerateAccessToken(f.userID, "organizer@test.invalid", []string{"organizer"})
|
||||
check(t, err)
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("GET /api/games/{id}/export/download", app.GameExportDownloadHandler(jwt, service))
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
request := func() *http.Request {
|
||||
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/api/games/%d/export/download", server.URL, f.gameID), nil)
|
||||
check(t, err)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
return req
|
||||
}
|
||||
interrupted, err := server.Client().Do(request())
|
||||
check(t, err)
|
||||
var prefix [16]byte
|
||||
_, err = io.ReadFull(interrupted.Body, prefix[:])
|
||||
check(t, err)
|
||||
check(t, interrupted.Body.Close())
|
||||
info, err := service.Get(context.Background(), f.gameID)
|
||||
check(t, err)
|
||||
if info.State != "ready" {
|
||||
t.Fatal("download abort changed export state")
|
||||
}
|
||||
check(t, f.teams.UpdateTeam(context.Background(), f.teamID, "Online edit after download abort"))
|
||||
response, err := server.Client().Do(request())
|
||||
check(t, err)
|
||||
defer func() { _ = response.Body.Close() }()
|
||||
data, err := io.ReadAll(response.Body)
|
||||
check(t, err)
|
||||
if response.StatusCode != http.StatusOK || !bytes.Equal(data, original) || response.Header.Get("Content-Type") != "application/zip" ||
|
||||
response.Header.Get("Content-Length") != fmt.Sprint(len(original)) {
|
||||
t.Fatal("streamed repeat download differs")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExportFailureUnlocksAndRetriesWithNewSnapshot(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
missingMedia, failedUpload bool
|
||||
}{
|
||||
{"unavailable media", true, false},
|
||||
{"partial upload", false, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newServiceFixture(t)
|
||||
storage := &memoryStorage{objects: map[string][]byte{}}
|
||||
if tc.failedUpload {
|
||||
storage.uploadErr = errors.New("private S3 endpoint secret detail")
|
||||
}
|
||||
worker := f.service(storage)
|
||||
if tc.missingMedia {
|
||||
worker = game_export_service.NewGameExportService(f.repo, f.codec, storage,
|
||||
func(context.Context, string) (io.ReadCloser, error) { return nil, errors.New("private media path") }, "https://game.example", false)
|
||||
}
|
||||
attempt, err := worker.Start(context.Background(), f.gameID, f.userID)
|
||||
check(t, err)
|
||||
worked, err := worker.ProcessNext(context.Background())
|
||||
if !worked || err == nil {
|
||||
t.Fatal("build failure ignored")
|
||||
}
|
||||
failed, err := worker.Get(context.Background(), f.gameID)
|
||||
check(t, err)
|
||||
if failed.State != "failed" || strings.Contains(failed.Failure, "private") || len(storage.objects) != 0 {
|
||||
t.Fatalf("unsafe or incomplete failure: %+v", failed)
|
||||
}
|
||||
|
||||
orphanDir, err := os.MkdirTemp("", "game-export-"+attempt.ID+"-")
|
||||
check(t, err)
|
||||
defer func() { _ = os.RemoveAll(orphanDir) }()
|
||||
check(t, os.WriteFile(filepath.Join(orphanDir, "partial.zip"), []byte("partial"), 0600))
|
||||
storage.objects[attempt.ID+".zip"] = []byte("leftover upload")
|
||||
check(t, worker.Maintain(context.Background()))
|
||||
if _, err := os.Stat(orphanDir); !errors.Is(err, os.ErrNotExist) || len(storage.objects) != 0 {
|
||||
t.Fatal("restart cleanup left temporary files or incomplete objects")
|
||||
}
|
||||
check(t, f.teams.UpdateTeam(context.Background(), f.teamID, "Changed for retry"))
|
||||
storage.uploadErr = nil
|
||||
worker = f.service(storage)
|
||||
retry, err := worker.Start(context.Background(), f.gameID, f.userID)
|
||||
check(t, err)
|
||||
if retry.ID == attempt.ID {
|
||||
t.Fatal("failed attempt reused")
|
||||
}
|
||||
worked, err = worker.ProcessNext(context.Background())
|
||||
check(t, err)
|
||||
if !worked {
|
||||
t.Fatal("retry not claimed")
|
||||
}
|
||||
_, data := download(t, worker, f.gameID)
|
||||
bundle, err := f.codec.Unpack(context.Background(), bytes.NewReader(data), int64(len(data)), t.TempDir())
|
||||
check(t, err)
|
||||
defer func() { _ = os.RemoveAll(bundle.Dir) }()
|
||||
if bundle.Data.Teams[0].Name != "Changed for retry" {
|
||||
t.Fatal("retry reused original snapshot")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type ambiguousCommitRepo struct {
|
||||
*game_exports_repo.GameExportsRepo
|
||||
}
|
||||
|
||||
func (r ambiguousCommitRepo) Complete(ctx context.Context, id, owner, packageID, digest string, size int64) error {
|
||||
if err := r.GameExportsRepo.Complete(ctx, id, owner, packageID, digest, size); err != nil {
|
||||
return err
|
||||
}
|
||||
return errors.New("connection lost after commit")
|
||||
}
|
||||
|
||||
func TestAmbiguousPublicationPreservesReadyObject(t *testing.T) {
|
||||
for _, tc := range []struct{ name string }{{"commit succeeds but connection fails"}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newServiceFixture(t)
|
||||
storage := &memoryStorage{objects: map[string][]byte{}}
|
||||
worker := game_export_service.NewGameExportService(ambiguousCommitRepo{f.repo}, f.codec, storage,
|
||||
func(context.Context, string) (io.ReadCloser, error) {
|
||||
return io.NopCloser(bytes.NewReader(f.media)), nil
|
||||
}, "https://game.example", false)
|
||||
_, err := worker.Start(context.Background(), f.gameID, f.userID)
|
||||
check(t, err)
|
||||
_, err = worker.ProcessNext(context.Background())
|
||||
if err == nil {
|
||||
t.Fatal("ambiguous commit ignored")
|
||||
}
|
||||
info, data := download(t, worker, f.gameID)
|
||||
if info.State != "ready" || len(data) == 0 || storage.deletes != 0 {
|
||||
t.Fatal("worker deleted published object")
|
||||
}
|
||||
check(t, worker.Maintain(context.Background()))
|
||||
_, repeated := download(t, worker, f.gameID)
|
||||
if !bytes.Equal(data, repeated) {
|
||||
t.Fatal("cleanup deleted ready object")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func verifyExportedGameReads(t *testing.T, f serviceFixture, worker *game_export_service.GameExportService, attemptID string) {
|
||||
t.Helper()
|
||||
games := game_service.NewGameService(f.games, f.teams, f.scenarios, nil, "", actions_repo.NewActionsRepo(f.pool), applications_repo.NewApplicationsRepo(f.pool))
|
||||
mapRepo := game_map_repo.NewGameMapRepo(f.pool)
|
||||
server := app.NewServer(nil, nil, nil, nil, games, nil, nil, nil, nil, nil, nil,
|
||||
screen_service.NewScreenService(screen_sessions_repo.NewScreenSessionsRepo(f.pool)), nil, nil, map_service.NewMapService(mapRepo), nil, worker)
|
||||
ctx := processor_jwt.ContextWithClaims(context.Background(), &processor_jwt.JWTClaims{UserID: f.userID, Roles: []string{"organizer"}})
|
||||
game, err := server.GetGame(ctx, &proto.GetGameReq{ID: int32(f.gameID)})
|
||||
check(t, err)
|
||||
if game.Game.Export == nil || game.Game.Export.ID != attemptID || game.Game.Export.State != "ready" {
|
||||
t.Fatal("game state lost export status")
|
||||
}
|
||||
items, err := server.GetGameMap(ctx, &proto.GetGameMapReq{GameID: int32(f.gameID)})
|
||||
check(t, err)
|
||||
stored, err := mapRepo.List(ctx, f.gameID)
|
||||
check(t, err)
|
||||
if len(stored) != 2 || len(items.Items) != 2 {
|
||||
t.Fatal("export status prevented normal map housekeeping")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package game_export_service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"evening_detective_server/internal/repos"
|
||||
"evening_detective_server/internal/repos/game_exports_repo"
|
||||
)
|
||||
|
||||
// GameExport is the public status shared with game reads.
|
||||
type GameExport = repos.GameExport
|
||||
|
||||
var (
|
||||
ErrGameNotFound = game_exports_repo.ErrGameNotFound
|
||||
ErrExportNotFound = game_exports_repo.ErrExportNotFound
|
||||
ErrNotExportable = game_exports_repo.ErrNotExportable
|
||||
)
|
||||
|
||||
// IGameExportsRepo is the worker's persistence boundary, including publication fencing.
|
||||
type IGameExportsRepo interface {
|
||||
Start(context.Context, int, int, string, string) (*game_exports_repo.Attempt, error)
|
||||
Get(context.Context, int) (*game_exports_repo.Attempt, error)
|
||||
Claim(context.Context, string) (*game_exports_repo.Attempt, error)
|
||||
Heartbeat(context.Context, string, string) error
|
||||
Complete(context.Context, string, string, string, string, int64) error
|
||||
Fail(context.Context, string, string, string) error
|
||||
Expire(context.Context) error
|
||||
CleanupKeys(context.Context) ([]string, error)
|
||||
MarkCleaned(context.Context, string) error
|
||||
IsInactive(context.Context, string) (bool, error)
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
package game_export_service_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/modules/file_storage"
|
||||
"github.com/kzzan/s3kit"
|
||||
)
|
||||
|
||||
func TestExportRustFSPrivateStreamingAndRestart(t *testing.T) {
|
||||
for _, tc := range []struct{ name string }{{"private ZIP survives storage and worker restart"}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
endpoint := os.Getenv("TEST_S3_ENDPOINT")
|
||||
if endpoint == "" {
|
||||
t.Skip("set TEST_S3_ENDPOINT, TEST_S3_USER and TEST_S3_PASSWORD for isolated RustFS checks")
|
||||
}
|
||||
user, password := os.Getenv("TEST_S3_USER"), os.Getenv("TEST_S3_PASSWORD")
|
||||
f := newServiceFixture(t)
|
||||
// An incompressible image makes the real HTTP abort happen during streaming.
|
||||
f.media = make([]byte, 2<<20)
|
||||
_, err := rand.Read(f.media)
|
||||
check(t, err)
|
||||
copy(f.media, []byte("\x89PNG\r\n\x1a\n"))
|
||||
bucket := fmt.Sprintf("ed029-export-test-%d", time.Now().UnixNano())
|
||||
ctx := context.Background()
|
||||
storage, err := file_storage.NewPrivateRustFSStorage(ctx, endpoint, user, password, bucket)
|
||||
check(t, err)
|
||||
client, err := s3kit.New(s3kit.Config{Endpoint: endpoint, AccessKeyID: user, SecretAccessKey: password, Region: "ru-1"})
|
||||
check(t, err)
|
||||
var key string
|
||||
t.Cleanup(func() {
|
||||
if key != "" {
|
||||
if err := storage.Delete(ctx, key); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
if err := client.DeleteBucket(ctx, bucket); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
})
|
||||
worker := f.service(storage)
|
||||
attempt, err := worker.Start(ctx, f.gameID, f.userID)
|
||||
check(t, err)
|
||||
key = attempt.ID + ".zip"
|
||||
worked, err := worker.ProcessNext(ctx)
|
||||
check(t, err)
|
||||
if !worked {
|
||||
t.Fatal("RustFS build not processed")
|
||||
}
|
||||
original := verifyPackage(t, f, worker)
|
||||
assertAbortedDownload(t, f, worker, original)
|
||||
anonymous, err := http.Get(endpoint + "/" + bucket + "/" + key)
|
||||
check(t, err)
|
||||
check(t, anonymous.Body.Close())
|
||||
if anonymous.StatusCode != http.StatusForbidden && anonymous.StatusCode != http.StatusNotFound {
|
||||
t.Fatalf("private package anonymously readable: %d", anonymous.StatusCode)
|
||||
}
|
||||
waitForStorageRestart(t)
|
||||
restartedStorage, err := file_storage.NewPrivateRustFSStorage(ctx, endpoint, user, password, bucket)
|
||||
check(t, err)
|
||||
restartedWorker := f.service(restartedStorage)
|
||||
_, repeated := download(t, restartedWorker, f.gameID)
|
||||
if !bytes.Equal(original, repeated) {
|
||||
t.Fatal("RustFS restart changed or lost package")
|
||||
}
|
||||
ready, err := restartedWorker.Start(ctx, f.gameID, f.userID)
|
||||
check(t, err)
|
||||
if ready.ID != attempt.ID || ready.State != "ready" {
|
||||
t.Fatal("restart rebuilt ready package")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The external test harness may restart only the isolated RustFS container while this schema stays alive.
|
||||
func waitForStorageRestart(t *testing.T) {
|
||||
t.Helper()
|
||||
marker := os.Getenv("TEST_S3_RESTART_MARKER")
|
||||
if marker == "" {
|
||||
return
|
||||
}
|
||||
_ = os.Remove(marker + ".continue")
|
||||
check(t, os.WriteFile(marker, []byte("ready"), 0600))
|
||||
defer func() { _ = os.Remove(marker); _ = os.Remove(marker + ".continue") }()
|
||||
deadline := time.Now().Add(3 * time.Minute)
|
||||
for time.Now().Before(deadline) {
|
||||
if _, err := os.Stat(marker + ".continue"); err == nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("isolated storage restart was not completed within three minutes")
|
||||
}
|
||||
@@ -0,0 +1,333 @@
|
||||
// Package game_export_service prepares durable, private game packages in online mode.
|
||||
package game_export_service
|
||||
|
||||
import (
|
||||
"archive/zip"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"evening_detective_server/internal/modules/file_storage"
|
||||
"evening_detective_server/internal/repos"
|
||||
"evening_detective_server/internal/repos/game_exports_repo"
|
||||
"evening_detective_server/internal/services/game_package_service"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrVenueMode = errors.New("экспорт игры доступен только на онлайн-сервере")
|
||||
ErrNotReady = errors.New("пакет игры ещё не готов")
|
||||
ErrInvalidGameID = errors.New("некорректный идентификатор игры")
|
||||
)
|
||||
|
||||
type GameExportService struct {
|
||||
repo IGameExportsRepo
|
||||
codec *game_package_service.GamePackageService
|
||||
storage file_storage.IStreamStorage
|
||||
openMedia game_package_service.OpenLocalFile
|
||||
source string
|
||||
venue bool
|
||||
}
|
||||
|
||||
func NewGameExportService(repo IGameExportsRepo, codec *game_package_service.GamePackageService,
|
||||
storage file_storage.IStreamStorage, openMedia game_package_service.OpenLocalFile, source string, venue bool) *GameExportService {
|
||||
return &GameExportService{
|
||||
repo: repo,
|
||||
codec: codec,
|
||||
storage: storage,
|
||||
openMedia: openMedia,
|
||||
source: source,
|
||||
venue: venue,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *GameExportService) Start(ctx context.Context, gameID, userID int) (*repos.GameExport, error) {
|
||||
if s.venue {
|
||||
return nil, ErrVenueMode
|
||||
}
|
||||
if gameID <= 0 {
|
||||
return nil, ErrInvalidGameID
|
||||
}
|
||||
id, err := file_storage.NewName()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
attempt, err := s.repo.Start(ctx, gameID, userID, id, s.source)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &attempt.GameExport, nil
|
||||
}
|
||||
|
||||
func (s *GameExportService) Get(ctx context.Context, gameID int) (*repos.GameExport, error) {
|
||||
if s.venue {
|
||||
return nil, ErrVenueMode
|
||||
}
|
||||
if gameID <= 0 {
|
||||
return nil, ErrInvalidGameID
|
||||
}
|
||||
attempt, err := s.repo.Get(ctx, gameID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &attempt.GameExport, nil
|
||||
}
|
||||
|
||||
// OpenDownload always opens the same stored bytes and never changes game or export state.
|
||||
func (s *GameExportService) OpenDownload(ctx context.Context, gameID int) (*repos.GameExport, io.ReadCloser, error) {
|
||||
if s.venue {
|
||||
return nil, nil, ErrVenueMode
|
||||
}
|
||||
if gameID <= 0 {
|
||||
return nil, nil, ErrInvalidGameID
|
||||
}
|
||||
attempt, err := s.repo.Get(ctx, gameID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if attempt.State != "ready" {
|
||||
return nil, nil, ErrNotReady
|
||||
}
|
||||
body, err := s.storage.Open(ctx, attempt.ObjectKey)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return &attempt.GameExport, body, nil
|
||||
}
|
||||
|
||||
// Run owns no in-memory queue. Restart recovery and cleanup are driven by PostgreSQL.
|
||||
func (s *GameExportService) Run(ctx context.Context) {
|
||||
if s.venue {
|
||||
return
|
||||
}
|
||||
maintenanceCtx, stop := context.WithCancel(ctx)
|
||||
defer stop()
|
||||
go func() {
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if err := s.Maintain(maintenanceCtx); err != nil && maintenanceCtx.Err() == nil {
|
||||
log.Printf("game export recovery: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-maintenanceCtx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}()
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for ctx.Err() == nil {
|
||||
worked, err := s.ProcessNext(ctx)
|
||||
if err != nil && ctx.Err() == nil {
|
||||
log.Printf("game export worker: %v", err)
|
||||
}
|
||||
if worked {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Maintain fails expired attempts before cleaning their private objects.
|
||||
func (s *GameExportService) Maintain(ctx context.Context) error {
|
||||
if s.venue {
|
||||
return ErrVenueMode
|
||||
}
|
||||
if err := s.repo.Expire(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
keys, err := s.repo.CleanupKeys(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, key := range keys {
|
||||
if err := s.storage.Delete(ctx, key); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.repo.MarkCleaned(ctx, key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return s.cleanTemporaryFiles(ctx)
|
||||
}
|
||||
|
||||
// ProcessNext processes at most one attempt; publication is fenced even after an upload succeeds.
|
||||
func (s *GameExportService) ProcessNext(ctx context.Context) (bool, error) {
|
||||
if s.venue {
|
||||
return false, ErrVenueMode
|
||||
}
|
||||
owner, err := file_storage.NewName()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
attempt, err := s.repo.Claim(ctx, owner)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
buildCtx, cancel := context.WithDeadline(ctx, attempt.CreatedAt.Add(game_exports_repo.PrepareTimeout))
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
ticker := time.NewTicker(30 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-buildCtx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if err := s.repo.Heartbeat(buildCtx, attempt.ID, owner); err != nil {
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
err = s.prepare(buildCtx, attempt, owner)
|
||||
cancel()
|
||||
<-done
|
||||
if err == nil {
|
||||
return true, nil
|
||||
}
|
||||
// Cleanup uses a fresh bounded context even when preparation was cancelled.
|
||||
cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cleanupCancel()
|
||||
failErr := s.repo.Fail(cleanupCtx, attempt.ID, owner, safeFailure(err))
|
||||
var deleteErr error
|
||||
if failErr == nil {
|
||||
deleteErr = s.storage.Delete(cleanupCtx, attempt.ObjectKey)
|
||||
}
|
||||
if errors.Is(failErr, game_exports_repo.ErrOwnershipLost) {
|
||||
failErr = nil
|
||||
}
|
||||
return true, errors.Join(err, failErr, deleteErr)
|
||||
}
|
||||
|
||||
func (s *GameExportService) prepare(ctx context.Context, attempt *game_exports_repo.Attempt, owner string) error {
|
||||
snapshot, err := mapSnapshot(attempt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dir, err := os.MkdirTemp("", "game-export-"+attempt.ID+"-")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = os.RemoveAll(dir) }()
|
||||
file, err := os.CreateTemp(dir, "archive-*.zip")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = file.Close(); _ = os.Remove(file.Name()) }()
|
||||
digest := sha256.New()
|
||||
if err := s.codec.PackInDirectory(ctx, snapshot, s.openMedia, io.MultiWriter(file, digest), dir); err != nil {
|
||||
return err
|
||||
}
|
||||
info, err := file.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
packageID, err := readPackageID(file, info.Size())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.repo.Heartbeat(ctx, attempt.ID, owner); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := file.Seek(0, io.SeekStart); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.storage.PutStream(ctx, attempt.ObjectKey, file, "application/zip"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
return s.repo.Complete(ctx, attempt.ID, owner, packageID, hex.EncodeToString(digest.Sum(nil)), info.Size())
|
||||
}
|
||||
|
||||
func readPackageID(file *os.File, size int64) (string, error) {
|
||||
archive, err := zip.NewReader(file, size)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
for _, entry := range archive.File {
|
||||
if entry.Name != "manifest.json" {
|
||||
continue
|
||||
}
|
||||
reader, err := entry.Open()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer func() { _ = reader.Close() }()
|
||||
var manifest game_package_service.Manifest
|
||||
if err := json.NewDecoder(io.LimitReader(reader, game_package_service.MaxJSONSize)).Decode(&manifest); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if manifest.PackageID == "" {
|
||||
return "", fmt.Errorf("empty package ID")
|
||||
}
|
||||
return manifest.PackageID, nil
|
||||
}
|
||||
return "", fmt.Errorf("package manifest missing")
|
||||
}
|
||||
|
||||
func safeFailure(err error) string {
|
||||
switch {
|
||||
case errors.Is(err, game_package_service.ErrMediaUnavailable), errors.Is(err, game_package_service.ErrUnsafeURL):
|
||||
return "Не удалось получить медиафайлы игры. Проверьте их доступность и повторите экспорт."
|
||||
case errors.Is(err, game_package_service.ErrLimitExceeded):
|
||||
return "Пакет игры превышает допустимый размер."
|
||||
case errors.Is(err, game_package_service.ErrInvalidPackage):
|
||||
return "Данные игры не подходят для автономного пакета."
|
||||
default:
|
||||
return "Не удалось подготовить пакет. Запустите экспорт повторно."
|
||||
}
|
||||
}
|
||||
|
||||
func (s *GameExportService) cleanTemporaryFiles(ctx context.Context) error {
|
||||
entries, err := os.ReadDir(os.TempDir())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, entry := range entries {
|
||||
name := entry.Name()
|
||||
if !entry.IsDir() || !strings.HasPrefix(name, "game-export-") {
|
||||
continue
|
||||
}
|
||||
suffix := strings.TrimPrefix(name, "game-export-")
|
||||
if len(suffix) < 34 || suffix[32] != '-' {
|
||||
continue
|
||||
}
|
||||
id := suffix[:32]
|
||||
if raw, err := hex.DecodeString(id); err != nil || len(raw) != 16 {
|
||||
continue
|
||||
}
|
||||
inactive, err := s.repo.IsInactive(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if inactive {
|
||||
if err := os.RemoveAll(filepath.Join(os.TempDir(), name)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
@@ -2,12 +2,14 @@
|
||||
package game_model
|
||||
|
||||
import (
|
||||
"evening_detective_server/internal/repos"
|
||||
"evening_detective_server/internal/services/scenarios_service"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Game describes a game exposed by the service layer.
|
||||
type Game struct {
|
||||
Export *repos.GameExport
|
||||
Format string
|
||||
ID int
|
||||
Name string
|
||||
|
||||
@@ -12,6 +12,7 @@ func MapGame(
|
||||
) *Game {
|
||||
return &Game{
|
||||
ID: o.ID,
|
||||
Export: o.Export,
|
||||
Name: o.Name,
|
||||
Description: o.Description,
|
||||
Location: o.Location,
|
||||
|
||||
@@ -51,6 +51,11 @@ func NewGamePackageService(fileDomain string) *GamePackageService {
|
||||
// Pack stages the complete archive before writing it. The caller discards output on error.
|
||||
// Snapshot IDs and secrets are preserved; the input snapshot is never mutated.
|
||||
func (s *GamePackageService) Pack(ctx context.Context, snapshot *Snapshot, openLocal OpenLocalFile, output io.Writer) error {
|
||||
return s.PackInDirectory(ctx, snapshot, openLocal, output, "")
|
||||
}
|
||||
|
||||
// PackInDirectory puts staging files under the export's owned temporary directory.
|
||||
func (s *GamePackageService) PackInDirectory(ctx context.Context, snapshot *Snapshot, openLocal OpenLocalFile, output io.Writer, tempRoot string) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -68,7 +73,7 @@ func (s *GamePackageService) Pack(ctx context.Context, snapshot *Snapshot, openL
|
||||
if err := validateData(data); err != nil {
|
||||
return err
|
||||
}
|
||||
dir, err := os.MkdirTemp("", "game-package-pack-")
|
||||
dir, err := os.MkdirTemp(tempRoot, "game-package-pack-")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -46,7 +46,8 @@ func TestServicesLifecyclePostgres(t *testing.T) {
|
||||
defer pool.Close()
|
||||
_, err = pool.Exec(ctx, `
|
||||
CREATE TEMP TABLE users(id INT, username TEXT);
|
||||
CREATE TEMP TABLE scenarios(id INT PRIMARY KEY, name TEXT, description TEXT, image TEXT, scenario TEXT,
|
||||
CREATE TEMP TABLE game_exports(id TEXT, game_id INT, state TEXT, package_id TEXT, size BIGINT, sha256 TEXT, failure TEXT, created_at TIMESTAMPTZ);
|
||||
CREATE TEMP TABLE scenarios(id INT PRIMARY KEY, playlist_id INT, name TEXT, description TEXT, image TEXT, scenario TEXT,
|
||||
author_id INT, status TEXT DEFAULT 'draft', updated_at TIMESTAMP DEFAULT NOW(), created_at TIMESTAMP DEFAULT NOW(), published_at TIMESTAMP, is_deleted BOOLEAN DEFAULT FALSE);
|
||||
CREATE TEMP TABLE games(id SERIAL PRIMARY KEY, name TEXT, description TEXT, location TEXT, start_at TIMESTAMP, scenario_id INT,
|
||||
format TEXT, status TEXT DEFAULT 'draft', started_at TIMESTAMP, ended_at TIMESTAMP, updated_at TIMESTAMP DEFAULT NOW(), created_at TIMESTAMP DEFAULT NOW());
|
||||
|
||||
@@ -30,6 +30,7 @@ func TestFinishedTeamStoryPostgres(t *testing.T) {
|
||||
}
|
||||
defer pool.Close()
|
||||
_, err = pool.Exec(ctx, `
|
||||
CREATE TEMP TABLE game_exports(id TEXT, game_id INT, state TEXT, package_id TEXT, size BIGINT, sha256 TEXT, failure TEXT, created_at TIMESTAMPTZ);
|
||||
CREATE TEMP TABLE games (id INT, name TEXT, description TEXT, location TEXT, start_at TIMESTAMP,
|
||||
format TEXT, status TEXT, scenario_id INT, started_at TIMESTAMP, ended_at TIMESTAMP);
|
||||
CREATE TEMP TABLE teams (id INT, name TEXT, captain_email TEXT, participants_count INT,
|
||||
@@ -55,7 +56,7 @@ func TestFinishedTeamStoryPostgres(t *testing.T) {
|
||||
t.Fatalf("finished game exposed content or lost status: story=%+v game=%+v actions=%+v", story, game, actions)
|
||||
}
|
||||
}
|
||||
if _, _, _, err := s.GetTeamActions(ctx, 1, "wrong"); !errors.Is(err, teams_repo.ErrTeamNotFound) {
|
||||
if _, _, _, err := s.GetTeamActions(ctx, 1, "wrong"); !errors.Is(err, ErrTeamNotFound) {
|
||||
t.Fatalf("password check bypassed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -78,7 +78,8 @@ func TestParticipationPostgres(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
exec(`CREATE TABLE games(id SERIAL PRIMARY KEY,name TEXT DEFAULT 'Game',start_at TIMESTAMP NOT NULL,status TEXT DEFAULT 'draft');
|
||||
exec(`CREATE TABLE game_exports(game_id INT, state TEXT);
|
||||
CREATE TABLE games(id SERIAL PRIMARY KEY,name TEXT DEFAULT 'Game',start_at TIMESTAMP NOT NULL,status TEXT DEFAULT 'draft');
|
||||
CREATE TABLE teams(id SERIAL PRIMARY KEY,name TEXT DEFAULT 'Team',creator_id INTEGER,game_id INTEGER REFERENCES games(id),password TEXT DEFAULT 'secret',status TEXT DEFAULT 'register',created_at TIMESTAMP DEFAULT (NOW() AT TIME ZONE 'UTC'));
|
||||
INSERT INTO games(start_at) VALUES((NOW() AT TIME ZONE 'UTC')+INTERVAL '12 hours'),((NOW() AT TIME ZONE 'UTC')+INTERVAL '2 days');
|
||||
INSERT INTO teams(game_id,created_at) VALUES(1,(NOW() AT TIME ZONE 'UTC')-INTERVAL '2 days'),(1,(NOW() AT TIME ZONE 'UTC')-INTERVAL '1 hour');`)
|
||||
@@ -221,14 +222,14 @@ func TestParticipationPostgres(t *testing.T) {
|
||||
}
|
||||
}
|
||||
gameplayService := gameplay_service.NewGameplayService(nil, repo, nil, nil, nil, nil, nil, "")
|
||||
if _, _, _, err := gameplayService.GetTeamActions(ctx, 3, "secret"); !errors.Is(err, teams_repo.ErrTeamNotFound) {
|
||||
if _, _, _, err := gameplayService.GetTeamActions(ctx, 3, "secret"); !errors.Is(err, gameplay_service.ErrTeamNotFound) {
|
||||
t.Fatalf("cancelled team got story: %v", err)
|
||||
}
|
||||
if err := gameplayService.AddTeamAction(ctx, 3, "secret", "anything", false); !errors.Is(err, teams_repo.ErrTeamNotFound) {
|
||||
if err := gameplayService.AddTeamAction(ctx, 3, "secret", "anything", false); !errors.Is(err, gameplay_service.ErrTeamNotFound) {
|
||||
t.Fatalf("cancelled team added action: %v", err)
|
||||
}
|
||||
answerService := answer_service.NewAnswerService(nil, repo, nil, nil, nil)
|
||||
if _, err := answerService.GetTeamAnswers(ctx, 3, "secret"); !errors.Is(err, teams_repo.ErrTeamNotFound) {
|
||||
if _, err := answerService.GetTeamAnswers(ctx, 3, "secret"); !errors.Is(err, answer_service.ErrTeamNotFound) {
|
||||
t.Fatalf("cancelled team got answers: %v", err)
|
||||
}
|
||||
token6 := participationToken("secret", 6)
|
||||
|
||||
Reference in New Issue
Block a user