diff --git a/internal/app/game_import.go b/internal/app/game_import.go new file mode 100644 index 0000000..413d6fe --- /dev/null +++ b/internal/app/game_import.go @@ -0,0 +1,107 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "log" + "mime" + "net/http" + "strings" + "time" + + "evening_detective_server/internal/modules/processor_jwt" + "evening_detective_server/internal/services/game_import_service" + "evening_detective_server/internal/services/game_package_service" + "evening_detective_server/proto" + "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" + "google.golang.org/genproto/googleapis/api/httpbody" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// IGameImporter is shared by the streaming HTTP and bounded gRPC transports. +type IGameImporter interface { + Import(context.Context, io.Reader, int) (*game_import_service.Result, error) +} + +func (s *server) ImportGame(ctx context.Context, req *httpbody.HttpBody) (*proto.GameImportRsp, error) { + claims, err := claimsFromContext(ctx) + if err != nil { + return nil, err + } + if !canExportGame(claims) { + return nil, status.Error(codes.PermissionDenied, "permission denied") + } + if !isZIP(req.GetContentType()) { + return nil, status.Error(codes.InvalidArgument, "Требуется application/zip") + } + result, err := s.gameImportService.Import(ctx, bytes.NewReader(req.GetData()), claims.UserID) + if err != nil { + return nil, gameImportErrorStatus(err) + } + return mapGameImport(result), nil +} + +// GameImportHandler receives ZIPs directly, without the gRPC message buffer or multipart encoding. +func GameImportHandler(jwt processor_jwt.IProcessorJWT, importer IGameImporter) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-Content-Type-Options", "nosniff") + fail := func(err error) { + code := runtime.HTTPStatusFromCode(status.Code(err)) + if status.Code(err) == codes.ResourceExhausted { + code = http.StatusRequestEntityTooLarge + } + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(map[string]string{"error": status.Convert(err).Message()}) + } + token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") + if !ok || 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 + } + if !isZIP(r.Header.Get("Content-Type")) { + fail(status.Error(codes.InvalidArgument, "Требуется application/zip")) + return + } + if r.ContentLength > game_package_service.MaxArchiveSize { + fail(gameImportErrorStatus(game_package_service.ErrLimitExceeded)) + return + } + if err := http.NewResponseController(w).SetReadDeadline(time.Now().Add(30 * time.Minute)); err != nil && !errors.Is(err, http.ErrNotSupported) { + fail(gameImportErrorStatus(err)) + return + } + r.Body = http.MaxBytesReader(w, r.Body, game_package_service.MaxArchiveSize) + result, err := importer.Import(r.Context(), r.Body, claims.UserID) + if err != nil { + fail(gameImportErrorStatus(err)) + return + } + if err := json.NewEncoder(w).Encode(map[string]any{ + "gameID": result.GameID, + "packageID": result.PackageID, + "repeated": result.Repeated, + }); err != nil { + log.Printf("game import response: %v", err) + } + }) +} + +func isZIP(contentType string) bool { + value, _, err := mime.ParseMediaType(contentType) + return err == nil && value == "application/zip" +} diff --git a/internal/app/game_import_errors.go b/internal/app/game_import_errors.go new file mode 100644 index 0000000..d324f86 --- /dev/null +++ b/internal/app/game_import_errors.go @@ -0,0 +1,37 @@ +package app + +import ( + "context" + "errors" + "log" + "net/http" + "os" + + "evening_detective_server/internal/services/game_import_service" + "evening_detective_server/internal/services/game_package_service" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func gameImportErrorStatus(err error) error { + var sizeError *http.MaxBytesError + switch { + case errors.Is(err, game_import_service.ErrOnlineMode): + return status.Error(codes.FailedPrecondition, "Импорт игры доступен только на площадке") + case errors.Is(err, game_import_service.ErrConflict): + return status.Error(codes.Aborted, "Пакет конфликтует с локальными данными. Существующие записи не изменены.") + case errors.Is(err, game_import_service.ErrInProgress), errors.Is(err, game_import_service.ErrInterrupted): + return status.Error(codes.Aborted, "Импорт выполняется или был прерван. Повторите загрузку позже.") + case errors.Is(err, game_package_service.ErrLimitExceeded), errors.As(err, &sizeError): + return status.Error(codes.ResourceExhausted, "Пакет игры превышает допустимый размер") + case errors.Is(err, game_package_service.ErrInvalidPackage), errors.Is(err, game_package_service.ErrUnsupportedVersion): + return status.Error(codes.InvalidArgument, "Некорректный или неподдерживаемый пакет игры") + case errors.Is(err, context.Canceled): + return status.Error(codes.Canceled, "Импорт отменён") + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, os.ErrDeadlineExceeded): + return status.Error(codes.DeadlineExceeded, "Время импорта истекло") + default: + log.Printf("game import: %v", err) + return status.Error(codes.Internal, "Не удалось импортировать игру") + } +} diff --git a/internal/app/game_import_mapper.go b/internal/app/game_import_mapper.go new file mode 100644 index 0000000..93b39c9 --- /dev/null +++ b/internal/app/game_import_mapper.go @@ -0,0 +1,14 @@ +package app + +import ( + "evening_detective_server/internal/services/game_import_service" + "evening_detective_server/proto" +) + +func mapGameImport(result *game_import_service.Result) *proto.GameImportRsp { + return &proto.GameImportRsp{ + GameID: int32(result.GameID), + PackageID: result.PackageID, + Repeated: result.Repeated, + } +} diff --git a/internal/app/game_import_test.go b/internal/app/game_import_test.go new file mode 100644 index 0000000..b9624e5 --- /dev/null +++ b/internal/app/game_import_test.go @@ -0,0 +1,170 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + + "evening_detective_server/internal/modules/processor_jwt" + "evening_detective_server/internal/services/game_import_service" + "evening_detective_server/internal/services/game_package_service" + "google.golang.org/genproto/googleapis/api/httpbody" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type importStub struct { + calls int + err error + actor int +} + +func (s *importStub) Import(ctx context.Context, body io.Reader, actor int) (*game_import_service.Result, error) { + s.calls++ + s.actor = actor + if s.err != nil { + return nil, s.err + } + raw, err := io.ReadAll(body) + if err != nil { + return nil, err + } + if string(raw) != "ZIP bytes" { + return nil, errors.New("wrong body") + } + return &game_import_service.Result{GameID: 100, PackageID: "package", Repeated: true}, nil +} + +func TestImportHTTPAccessAndResponse(t *testing.T) { + for _, tc := range []struct { + name, role, auth, mime string + size int64 + want int + called bool + }{ + {"organizer", "organizer", "valid", "application/zip", 9, 200, true}, + {"admin", "admin", "valid", "application/zip", 9, 200, true}, + {"developer", "developer", "valid", "application/zip", 9, 200, true}, + {"author", "author", "valid", "application/zip", 9, 403, false}, + {"user", "user", "valid", "application/zip", 9, 403, false}, + {"missing JWT", "", "", "application/zip", 9, 401, false}, + {"bad JWT", "", "Bearer bad", "application/zip", 9, 401, false}, + {"missing bearer prefix", "organizer", "no-prefix", "application/zip", 9, 401, false}, + {"wrong MIME", "organizer", "valid", "application/json", 9, 400, false}, + {"oversized", "organizer", "valid", "application/zip", game_package_service.MaxArchiveSize + 1, 413, false}, + {"chunked upload", "organizer", "valid", "application/zip", -1, 200, true}, + } { + t.Run(tc.name, func(t *testing.T) { + jwt := processor_jwt.NewProcessor("import-test-secret") + token, err := jwt.GenerateAccessToken(17, "organizer@test.invalid", []string{tc.role}) + if err != nil { + t.Fatal(err) + } + stub := &importStub{} + req := httptest.NewRequest(http.MethodPost, "/api/games/import", strings.NewReader("ZIP bytes")) + req.ContentLength = tc.size + req.Header.Set("Content-Type", tc.mime) + switch tc.auth { + case "valid": + req.Header.Set("Authorization", "Bearer "+token) + case "no-prefix": + req.Header.Set("Authorization", token) + default: + req.Header.Set("Authorization", tc.auth) + } + w := httptest.NewRecorder() + GameImportHandler(jwt, stub).ServeHTTP(w, req) + if w.Code != tc.want || (stub.calls == 1) != tc.called { + t.Fatalf("status %d, calls %d", w.Code, stub.calls) + } + if w.Header().Get("Cache-Control") != "no-store" { + t.Fatal("response cacheable") + } + if tc.called { + var result struct { + GameID int `json:"gameID"` + PackageID string `json:"packageID"` + Repeated bool `json:"repeated"` + } + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + if stub.actor != 17 || result.GameID != 100 || result.PackageID != "package" || !result.Repeated { + t.Fatal("incorrect response or actor") + } + } + }) + } +} + +func TestImportErrorMapping(t *testing.T) { + for _, tc := range []struct { + name string + err error + code codes.Code + http int + }{ + {"online", game_import_service.ErrOnlineMode, codes.FailedPrecondition, 400}, + {"conflict", game_import_service.ErrConflict, codes.Aborted, 409}, + {"busy", game_import_service.ErrInProgress, codes.Aborted, 409}, + {"interrupted", game_import_service.ErrInterrupted, codes.Aborted, 409}, + {"invalid", game_package_service.ErrInvalidPackage, codes.InvalidArgument, 400}, + {"version", game_package_service.ErrUnsupportedVersion, codes.InvalidArgument, 400}, + {"limit", game_package_service.ErrLimitExceeded, codes.ResourceExhausted, 413}, + {"body limit", &http.MaxBytesError{Limit: 1}, codes.ResourceExhausted, 413}, + {"cancel", context.Canceled, codes.Canceled, 499}, + {"deadline", context.DeadlineExceeded, codes.DeadlineExceeded, 504}, + {"read deadline", os.ErrDeadlineExceeded, codes.DeadlineExceeded, 504}, + {"unknown", errors.New("private SQL secret path"), codes.Internal, 500}, + } { + t.Run(tc.name, func(t *testing.T) { + mapped := gameImportErrorStatus(fmt.Errorf("private detail: %w", tc.err)) + if status.Code(mapped) != tc.code || strings.Contains(status.Convert(mapped).Message(), "private") { + t.Fatalf("unsafe or wrong mapping: %v", mapped) + } + jwt := processor_jwt.NewProcessor("import-test-secret") + token, err := jwt.GenerateAccessToken(1, "test@invalid", []string{"organizer"}) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, "/api/games/import", strings.NewReader("ZIP bytes")) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/zip") + w := httptest.NewRecorder() + GameImportHandler(jwt, &importStub{err: tc.err}).ServeHTTP(w, req) + if w.Code != tc.http || strings.Contains(w.Body.String(), "private") { + t.Fatalf("unsafe or wrong HTTP response: %d %s", w.Code, w.Body.String()) + } + }) + } +} + +func TestImportRPCRejectsUnauthorizedCallers(t *testing.T) { + for _, tc := range []struct { + name, role, mime string + want codes.Code + }{ + {"guest", "", "application/zip", codes.Unauthenticated}, + {"user", "user", "application/zip", codes.PermissionDenied}, + {"author", "author", "application/zip", codes.PermissionDenied}, + {"invalid content type", "organizer", "text/plain", codes.InvalidArgument}, + } { + t.Run(tc.name, 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{}).ImportGame(ctx, &httpbody.HttpBody{ContentType: tc.mime}) + if status.Code(err) != tc.want { + t.Fatalf("got %v", err) + } + }) + } +} diff --git a/internal/app/server.go b/internal/app/server.go index 6a3ef46..5f54318 100644 --- a/internal/app/server.go +++ b/internal/app/server.go @@ -4,6 +4,7 @@ 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_import_service" "evening_detective_server/internal/services/game_review_service" "evening_detective_server/internal/services/game_service" "evening_detective_server/internal/services/gameplay_service" @@ -30,6 +31,7 @@ type server struct { fileService *file_service.FileService scenarioService *scenarios_service.ScenarioService gameExportService *game_export_service.GameExportService + gameImportService *game_import_service.GameImportService gameService *game_service.GameService gameplayService *gameplay_service.GameplayService answerService *answer_service.AnswerService @@ -62,8 +64,10 @@ func NewServer( mapService *map_service.MapService, taskService *task_service.TaskService, gameExportService *game_export_service.GameExportService, + gameImportService *game_import_service.GameImportService, ) proto.EveningDetectiveServerServer { return &server{ + gameImportService: gameImportService, taskService: taskService, gameExportService: gameExportService, usersService: usersService, diff --git a/internal/repos/game_imports_repo/publish.go b/internal/repos/game_imports_repo/publish.go new file mode 100644 index 0000000..5a6f6d3 --- /dev/null +++ b/internal/repos/game_imports_repo/publish.go @@ -0,0 +1,352 @@ +package game_imports_repo + +import ( + "context" + "encoding/json" + + "github.com/jackc/pgx/v5" +) + +// Complete makes all game rows visible together, after media has been uploaded. +func (r *GameImportsRepo) Complete(ctx context.Context, id string, data []byte) error { + tx, err := r.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + var actor int + var live bool + err = tx.QueryRow( + ctx, + `SELECT + COALESCE(initiator_id, 0), + state = 'preparing' AND lease_until > clock_timestamp() + AND created_at + $2::interval > clock_timestamp() + FROM game_imports + WHERE id = $1 + FOR UPDATE`, + id, + r.prepareTimeout.String(), + ).Scan(&actor, &live) + if err != nil { + return err + } + if !live || actor == 0 { + return ErrOwnershipLost + } + // Block concurrent INSERTs before inspecting IDs and advancing their sequences. + _, err = tx.Exec( + ctx, + `LOCK TABLE games, teams, tickets IN SHARE ROW EXCLUSIVE MODE`, + ) + if err != nil { + return err + } + if err := checkConflicts(ctx, tx, data); err != nil { + return err + } + mappings := Mappings{ + Playlists: map[int]int{}, + Tracks: map[int]int{}, + } + if err := insertPlaylists(ctx, tx, data, &mappings); err != nil { + return storageError(err) + } + playlistJSON, err := json.Marshal(mappings.Playlists) + if err != nil { + return err + } + err = tx.QueryRow( + ctx, + `INSERT INTO scenarios ( + name, description, image, scenario, author_id, status, + created_at, updated_at, published_at, playlist_id + ) + SELECT + s.name, + s.description, + s.image, + s.story, + $2, + s.status, + s.created_at AT TIME ZONE 'UTC', + s.updated_at AT TIME ZONE 'UTC', + s.published_at AT TIME ZONE 'UTC', + ($3::jsonb->>s.playlist_id::text)::integer + FROM jsonb_to_record($1::jsonb->'scenario') AS s( + name text, description text, image text, story jsonb, status text, + created_at timestamptz, updated_at timestamptz, published_at timestamptz, playlist_id integer + ) + RETURNING + id`, + data, + actor, + playlistJSON, + ).Scan(&mappings.ScenarioID) + if err != nil { + return storageError(err) + } + _, err = tx.Exec( + ctx, + `INSERT INTO games ( + id, name, description, location, start_at, status, format, scenario_id, playlist_id + ) + SELECT + g.id, + g.name, + g.description, + g.location, + g.start_at AT TIME ZONE 'UTC', + g.status, + g.format, + $2, + ($3::jsonb->>g.playlist_id::text)::integer + FROM jsonb_to_record($1::jsonb->'game') AS g( + id integer, name text, description text, location text, start_at timestamptz, + status text, format text, playlist_id integer + )`, + data, + mappings.ScenarioID, + playlistJSON, + ) + if err != nil { + return storageError(err) + } + if err := insertGameRelations(ctx, tx, data); err != nil { + return storageError(err) + } + // setval is not transactional; a rollback may leave a gap but cannot reuse an ID. + _, err = tx.Exec( + ctx, + `SELECT + setval('games_id_seq', GREATEST( + ( + SELECT + COALESCE(MAX(id), 1) + FROM games + ), + ( + SELECT + last_value + FROM games_id_seq + ) + )), + setval('teams_id_seq', GREATEST( + ( + SELECT + COALESCE(MAX(id), 1) + FROM teams + ), + ( + SELECT + last_value + FROM teams_id_seq + ) + )), + setval('tickets_id_seq', GREATEST( + ( + SELECT + COALESCE(MAX(id), 1) + FROM tickets + ), + ( + SELECT + last_value + FROM tickets_id_seq + ) + ))`, + ) + if err != nil { + return err + } + raw, err := json.Marshal(mappings) + if err != nil { + return err + } + tag, err := tx.Exec( + ctx, + `UPDATE game_imports + SET state = 'ready', mappings = $2 + WHERE id = $1 AND state = 'preparing' AND lease_until > clock_timestamp() + AND created_at + $3::interval > clock_timestamp()`, + id, + raw, + r.prepareTimeout.String(), + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrOwnershipLost + } + return tx.Commit(ctx) +} + +func insertPlaylists(ctx context.Context, tx pgx.Tx, data []byte, mappings *Mappings) error { + rows, err := tx.Query( + ctx, + `SELECT + id, + name, + tracks + FROM jsonb_to_recordset(COALESCE(NULLIF($1::jsonb->'playlists', 'null'::jsonb), '[]'::jsonb)) AS p(id integer, name text, tracks jsonb) + ORDER BY id`, + data, + ) + if err != nil { + return err + } + defer rows.Close() + type playlist struct { + id int + name string + tracks []byte + } + var playlists []playlist + for rows.Next() { + var p playlist + if err := rows.Scan(&p.id, &p.name, &p.tracks); err != nil { + return err + } + playlists = append(playlists, p) + } + if err := rows.Err(); err != nil { + return err + } + rows.Close() + for _, p := range playlists { + var localID int + err := tx.QueryRow( + ctx, + `INSERT INTO playlists (name) + VALUES ($1) + RETURNING + id`, + p.name, + ).Scan(&localID) + if err != nil { + return err + } + mappings.Playlists[p.id] = localID + var trackMap []byte + err = tx.QueryRow( + ctx, + `WITH source AS MATERIALIZED ( + SELECT + id AS source_id, + nextval(pg_get_serial_sequence('playlist_tracks', 'id'))::integer AS local_id, + name, + filename, + position + FROM jsonb_to_recordset(COALESCE(NULLIF($1::jsonb, 'null'::jsonb), '[]'::jsonb)) AS t(id integer, name text, filename text, position integer) + ), inserted AS ( + INSERT INTO playlist_tracks (id, playlist_id, name, filename, position) + SELECT + local_id, + $2, + name, + filename, + position + FROM source + RETURNING + id + ) + SELECT + COALESCE(jsonb_object_agg(s.source_id, s.local_id), '{}'::jsonb) + FROM source s + JOIN inserted i ON i.id = s.local_id`, + p.tracks, + localID, + ).Scan(&trackMap) + if err != nil { + return err + } + if err := json.Unmarshal(trackMap, &mappings.Tracks); err != nil { + return err + } + } + return nil +} + +func insertGameRelations(ctx context.Context, tx pgx.Tx, data []byte) error { + _, err := tx.Exec( + ctx, + `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 + t.id, + t.game_id, + t.name, + t.status, + t.password, + NULL, + t.captain_email, + t.participants_count, + t.participation_status, + t.participation_source, + t.participation_answered_at, + t.registration_delivery_id, + t.created_at AT TIME ZONE 'UTC' + FROM jsonb_to_recordset(COALESCE(NULLIF($1::jsonb->'teams', 'null'::jsonb), '[]'::jsonb)) AS t( + id integer, game_id integer, name text, status text, password text, captain_email text, + participants_count integer, participation_status text, participation_source text, + participation_answered_at timestamptz, registration_delivery_id text, created_at timestamptz + )`, + data, + ) + if err != nil { + return err + } + _, err = tx.Exec( + ctx, + `INSERT INTO tickets (id, game_id, token_hash, status, used_at, used_by, created_at) + SELECT + t.id, + t.game_id, + decode(t.token_hash, 'base64'), + t.status, + t.used_at, + NULL, + t.created_at + FROM jsonb_to_recordset(COALESCE(NULLIF($1::jsonb->'tickets', 'null'::jsonb), '[]'::jsonb)) AS t( + id integer, game_id integer, token_hash text, status text, used_at timestamptz, created_at timestamptz + )`, + data, + ) + if err != nil { + return err + } + _, err = tx.Exec( + ctx, + `INSERT INTO game_screen_plans (game_id, steps, current_step) + SELECT + p.game_id, + p.steps, + p.current_step + FROM jsonb_to_record(COALESCE(NULLIF($1::jsonb->'screen_plan', 'null'::jsonb), '{}'::jsonb)) AS p(game_id integer, steps jsonb, current_step integer) + WHERE p.game_id IS NOT NULL`, + data, + ) + if err != nil { + return err + } + _, err = tx.Exec( + ctx, + `INSERT INTO game_map_items (game_id, kind, item_id, x, y, hidden) + SELECT + m.game_id, + m.kind, + m.item_id, + m.x, + m.y, + m.hidden + FROM jsonb_to_recordset(COALESCE(NULLIF($1::jsonb->'map', 'null'::jsonb), '[]'::jsonb)) AS m( + game_id integer, kind text, item_id integer, x integer, y integer, hidden boolean + )`, + data, + ) + return err +} diff --git a/internal/repos/game_imports_repo/repo.go b/internal/repos/game_imports_repo/repo.go new file mode 100644 index 0000000..bc17e95 --- /dev/null +++ b/internal/repos/game_imports_repo/repo.go @@ -0,0 +1,329 @@ +// Package game_imports_repo stages imports and atomically publishes local game data. +package game_imports_repo + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ( + ErrConflict = errors.New("пакет или его идентификаторы конфликтуют с локальными данными") + ErrInProgress = errors.New("импорт этой игры уже выполняется") + ErrOwnershipLost = errors.New("владение импортом потеряно") + ErrInvalidData = errors.New("данные пакета не соответствуют схеме площадки") +) + +const Lease = 2 * time.Minute +const PrepareTimeout = 30 * time.Minute + +// Attempt contains private recovery data, never a public API response. +type Attempt struct { + Mappings Mappings + ID string + PackageID string + SourceServer string + GameID int + SHA256 string + InitiatorID int + Manifest []byte + OriginalData []byte + ObjectKeys []string + State string + CreatedAt time.Time +} + +// Mappings retains the source IDs of locally allocated copies. +type Mappings struct { + ScenarioID int `json:"scenario_id"` + Playlists map[int]int `json:"playlists"` + Tracks map[int]int `json:"tracks"` +} + +type GameImportsRepo struct { + pool *pgxpool.Pool + lease time.Duration + prepareTimeout time.Duration +} + +func NewGameImportsRepo(pool *pgxpool.Pool) *GameImportsRepo { + return &GameImportsRepo{ + pool: pool, + lease: Lease, + prepareTimeout: PrepareTimeout, + } +} + +// Start reserves IDs before any media is uploaded. Failed attempts may be retried. +func (r *GameImportsRepo) Start(ctx context.Context, a *Attempt) (*Attempt, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback(ctx) }() + tag, err := tx.Exec( + ctx, + `INSERT INTO game_imports ( + id, package_id, source_server, source_game_id, sha256, + initiator_id, manifest, original_data, object_keys, lease_until + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, clock_timestamp() + $10::interval) + ON CONFLICT DO NOTHING`, + a.ID, + a.PackageID, + a.SourceServer, + a.GameID, + a.SHA256, + a.InitiatorID, + a.Manifest, + a.OriginalData, + a.ObjectKeys, + r.lease.String(), + ) + if err != nil { + return nil, storageError(err) + } + if tag.RowsAffected() == 0 { + existing := &Attempt{} + err := tx.QueryRow( + ctx, + `SELECT + id, + package_id, + source_server, + source_game_id, + sha256, + state, + created_at + FROM game_imports + WHERE state IN ('preparing', 'ready') + AND (package_id = $1 OR source_game_id = $2) + ORDER BY package_id = $1 DESC + LIMIT 1`, + a.PackageID, + a.GameID, + ).Scan(&existing.ID, &existing.PackageID, &existing.SourceServer, &existing.GameID, + &existing.SHA256, &existing.State, &existing.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrConflict + } + if err != nil { + return nil, err + } + if existing.PackageID != a.PackageID || existing.SHA256 != a.SHA256 || + existing.SourceServer != a.SourceServer || existing.GameID != a.GameID { + return nil, ErrConflict + } + if existing.State != "ready" { + return nil, ErrInProgress + } + return existing, tx.Commit(ctx) + } + if err := checkConflicts(ctx, tx, a.OriginalData); err != nil { + return nil, err + } + err = tx.QueryRow( + ctx, + `SELECT + state, + created_at + FROM game_imports + WHERE id = $1`, + a.ID, + ).Scan(&a.State, &a.CreatedAt) + if err != nil { + return nil, err + } + return a, tx.Commit(ctx) +} + +func checkConflicts(ctx context.Context, tx pgx.Tx, raw []byte) error { + var conflict bool + err := tx.QueryRow( + ctx, + `SELECT + EXISTS ( + SELECT + 1 + FROM games + WHERE id = ($1::jsonb->'game'->>'id')::integer + ) OR EXISTS ( + SELECT + 1 + FROM teams t + JOIN jsonb_to_recordset(COALESCE(NULLIF($1::jsonb->'teams', 'null'::jsonb), '[]'::jsonb)) AS s(id integer) ON s.id = t.id + ) OR EXISTS ( + SELECT + 1 + FROM tickets t + JOIN jsonb_to_recordset(COALESCE(NULLIF($1::jsonb->'tickets', 'null'::jsonb), '[]'::jsonb)) AS s(id integer, token_hash text) + ON t.id = s.id OR t.token_hash = decode(s.token_hash, 'base64') + )`, + raw, + ).Scan(&conflict) + if err != nil { + return storageError(err) + } + if conflict { + return ErrConflict + } + return nil +} + +func (r *GameImportsRepo) Heartbeat(ctx context.Context, id string) error { + tag, err := r.pool.Exec( + ctx, + `UPDATE game_imports + SET lease_until = clock_timestamp() + $2::interval + WHERE id = $1 AND state = 'preparing' AND lease_until > clock_timestamp() + AND created_at + $3::interval > clock_timestamp()`, + id, + r.lease.String(), + r.prepareTimeout.String(), + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrOwnershipLost + } + return nil +} + +// Fail cannot invalidate a committed game, including after an ambiguous commit response. +func (r *GameImportsRepo) Fail(ctx context.Context, id string) error { + tag, err := r.pool.Exec( + ctx, + `UPDATE game_imports + SET state = 'failed' + WHERE id = $1 AND state = 'preparing'`, + id, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrOwnershipLost + } + return nil +} + +func (r *GameImportsRepo) Expire(ctx context.Context) error { + _, err := r.pool.Exec( + ctx, + `UPDATE game_imports + SET state = 'failed' + WHERE state = 'preparing' AND ( + lease_until <= clock_timestamp() OR created_at + $1::interval <= clock_timestamp() + )`, + r.prepareTimeout.String(), + ) + return err +} + +// CleanupAttempts keeps retrying long enough to remove uploads from stale workers. +func (r *GameImportsRepo) CleanupAttempts(ctx context.Context) ([]Attempt, error) { + rows, err := r.pool.Query( + ctx, + `SELECT + id, + object_keys + FROM game_imports + 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() + var attempts []Attempt + for rows.Next() { + var a Attempt + if err := rows.Scan(&a.ID, &a.ObjectKeys); err != nil { + return nil, err + } + attempts = append(attempts, a) + } + return attempts, rows.Err() +} + +func (r *GameImportsRepo) MarkCleaned(ctx context.Context, id string) error { + _, err := r.pool.Exec( + ctx, + `UPDATE game_imports + SET cleaned_at = clock_timestamp() + WHERE id = $1 AND state = 'failed'`, + id, + ) + return err +} + +// IsInactive also treats an unregistered attempt as inactive after its upload deadline. +func (r *GameImportsRepo) IsInactive(ctx context.Context, id string) (bool, error) { + var inactive bool + err := r.pool.QueryRow( + ctx, + `SELECT + NOT EXISTS ( + SELECT + 1 + FROM game_imports + WHERE id = $1 AND state = 'preparing' + )`, + id, + ).Scan(&inactive) + return inactive, err +} + +func storageError(err error) error { + var postgres *pgconn.PgError + if errors.As(err, &postgres) { + switch postgres.Code { + case "23505": + return errors.Join(ErrConflict, err) + case "22001", "22003", "23514", "22007", "22008": + return errors.Join(ErrInvalidData, err) + } + } + return err +} + +// Get returns private metadata for recovery and the later results-return stage. +func (r *GameImportsRepo) Get(ctx context.Context, packageID string) (*Attempt, error) { + attempt := &Attempt{} + err := r.pool.QueryRow( + ctx, + `SELECT + id, + package_id, + source_server, + source_game_id, + sha256, + COALESCE(initiator_id, 0), + manifest, + original_data, + mappings, + object_keys, + state, + created_at + FROM game_imports + WHERE package_id = $1 + ORDER BY created_at DESC, id + LIMIT 1`, + packageID, + ).Scan(&attempt.ID, &attempt.PackageID, &attempt.SourceServer, &attempt.GameID, + &attempt.SHA256, &attempt.InitiatorID, &attempt.Manifest, &attempt.OriginalData, + &attempt.Mappings, &attempt.ObjectKeys, &attempt.State, &attempt.CreatedAt) + return attempt, err +} diff --git a/internal/repos/game_imports_repo/repo_test.go b/internal/repos/game_imports_repo/repo_test.go new file mode 100644 index 0000000..fe2ccc7 --- /dev/null +++ b/internal/repos/game_imports_repo/repo_test.go @@ -0,0 +1,131 @@ +package game_imports_repo + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "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" +) + +// repoFixture prepares only schema infrastructure; records use public repositories. +func repoFixture(t *testing.T) (*GameImportsRepo, int) { + t.Helper() + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + t.Skip("set TEST_DATABASE_URL for import lease checks") + } + ctx := context.Background() + admin, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + schema := fmt.Sprintf("import_lease_test_%d", time.Now().UnixNano()) + quoted := pgx.Identifier{schema}.Sanitize() + if _, err := admin.Exec(ctx, "CREATE SCHEMA "+quoted); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if _, err := admin.Exec(ctx, "DROP SCHEMA "+quoted+" CASCADE"); err != nil { + t.Error(err) + } + admin.Close() + }) + config, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + config.ConnConfig.RuntimeParams["search_path"] = schema + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + names, err := filepath.Glob("../../../migrations/*.sql") + if err != nil { + t.Fatal(err) + } + for _, name := range names { + raw, err := os.ReadFile(name) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, strings.Split(string(raw), "-- +goose Down")[0]); err != nil { + t.Fatalf("migration %s: %v", name, err) + } + } + actor, err := users_repo.NewUserRepo(pool).AddUserWithAgreements(ctx, "venue", "lease@test.invalid", "hash", []string{"organizer"}, []user_agreements_repo.Agreement{{AgreementType: "terms", Version: "test"}}) + if err != nil { + t.Fatal(err) + } + return NewGameImportsRepo(pool), actor +} + +func TestExpiredImportsCannotPublishAndKeepCleanupKeys(t *testing.T) { + for _, tc := range []struct { + name string + lease, timeout time.Duration + }{ + {"expired lease", -time.Second, PrepareTimeout}, + {"expired total timeout", Lease, -time.Second}, + } { + t.Run(tc.name, func(t *testing.T) { + repo, actor := repoFixture(t) + repo.lease, repo.prepareTimeout = tc.lease, tc.timeout + ctx := context.Background() + data := []byte(`{"game":{"id":100},"teams":[],"tickets":[]}`) + attempt, err := repo.Start(ctx, &Attempt{ + ID: "expired-attempt", + PackageID: "expired-package", + SourceServer: "https://online.example", + GameID: 100, + SHA256: "digest", + InitiatorID: actor, + Manifest: []byte(`{}`), + OriginalData: data, + ObjectKeys: []string{"partial-upload"}, + }) + if err != nil { + t.Fatal(err) + } + if err := repo.Heartbeat(ctx, attempt.ID); !errors.Is(err, ErrOwnershipLost) { + t.Fatalf("expired heartbeat: %v", err) + } + if err := repo.Complete(ctx, attempt.ID, data); !errors.Is(err, ErrOwnershipLost) { + t.Fatalf("expired publication: %v", err) + } + if err := repo.Expire(ctx); err != nil { + t.Fatal(err) + } + metadata, err := repo.Get(ctx, attempt.PackageID) + if err != nil { + t.Fatal(err) + } + if metadata.State != "failed" { + t.Fatal("expired attempt not failed") + } + keys, err := repo.CleanupAttempts(ctx) + if err != nil { + t.Fatal(err) + } + if len(keys) != 1 || keys[0].ObjectKeys[0] != "partial-upload" { + t.Fatal("cleanup keys lost") + } + if err := repo.MarkCleaned(ctx, attempt.ID); err != nil { + t.Fatal(err) + } + inactive, err := repo.IsInactive(ctx, attempt.ID) + if err != nil || !inactive { + t.Fatal("expired attempt still active") + } + }) + } +} diff --git a/internal/services/game_export_service/export_test.go b/internal/services/game_export_service/export_test.go index 0def0d6..62bc42e 100644 --- a/internal/services/game_export_service/export_test.go +++ b/internal/services/game_export_service/export_test.go @@ -377,7 +377,7 @@ func verifyExportedGameReads(t *testing.T, f serviceFixture, worker *game_export 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) + screen_service.NewScreenService(screen_sessions_repo.NewScreenSessionsRepo(f.pool)), nil, nil, map_service.NewMapService(mapRepo), nil, worker, nil) 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) diff --git a/internal/services/game_import_service/import_errors.go b/internal/services/game_import_service/import_errors.go new file mode 100644 index 0000000..613533a --- /dev/null +++ b/internal/services/game_import_service/import_errors.go @@ -0,0 +1,29 @@ +package game_import_service + +import ( + "errors" + + "evening_detective_server/internal/repos/game_imports_repo" + "evening_detective_server/internal/services/game_package_service" +) + +var ( + ErrConflict = errors.New("пакет конфликтует с локальными данными") + ErrInProgress = errors.New("импорт этой игры уже выполняется") + ErrInterrupted = errors.New("импорт прерван, повторите загрузку") +) + +func importError(err error) error { + switch { + case errors.Is(err, game_imports_repo.ErrConflict): + return errors.Join(ErrConflict, err) + case errors.Is(err, game_imports_repo.ErrInProgress): + return errors.Join(ErrInProgress, err) + case errors.Is(err, game_imports_repo.ErrOwnershipLost): + return errors.Join(ErrInterrupted, err) + case errors.Is(err, game_imports_repo.ErrInvalidData): + return errors.Join(game_package_service.ErrInvalidPackage, err) + default: + return err + } +} diff --git a/internal/services/game_import_service/import_mapper.go b/internal/services/game_import_service/import_mapper.go new file mode 100644 index 0000000..c2e924d --- /dev/null +++ b/internal/services/game_import_service/import_mapper.go @@ -0,0 +1,11 @@ +package game_import_service + +import "evening_detective_server/internal/repos/game_imports_repo" + +func mapResult(attempt *game_imports_repo.Attempt, repeated bool) *Result { + return &Result{ + GameID: attempt.GameID, + PackageID: attempt.PackageID, + Repeated: repeated, + } +} diff --git a/internal/services/game_import_service/import_test.go b/internal/services/game_import_service/import_test.go new file mode 100644 index 0000000..1e87031 --- /dev/null +++ b/internal/services/game_import_service/import_test.go @@ -0,0 +1,707 @@ +package game_import_service_test + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "evening_detective_server/internal/modules/cleaner" + "evening_detective_server/internal/modules/file_storage" + "evening_detective_server/internal/modules/storytelling" + "evening_detective_server/internal/repos/actions_repo" + "evening_detective_server/internal/repos/applications_repo" + "evening_detective_server/internal/repos/game_imports_repo" + "evening_detective_server/internal/repos/game_map_repo" + "evening_detective_server/internal/repos/game_screen_plans_repo" + "evening_detective_server/internal/repos/games_repo" + "evening_detective_server/internal/repos/playlists_repo" + "evening_detective_server/internal/repos/scenarios_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" + imports "evening_detective_server/internal/services/game_import_service" + "evening_detective_server/internal/services/game_package_service" + "evening_detective_server/internal/services/gameplay_service" + "evening_detective_server/internal/services/scenarios_service" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type memoryStorage struct { + mu sync.Mutex + objects map[string][]byte + uploadErr error + deleteErr error + started chan struct{} + release chan struct{} +} + +func (s *memoryStorage) PutStream(ctx context.Context, key string, reader io.Reader, _ string) error { + if s.started != nil { + select { + case s.started <- struct{}{}: + default: + } + select { + case <-s.release: + case <-ctx.Done(): + return ctx.Err() + } + } + body, err := io.ReadAll(reader) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + s.objects[key] = body + return s.uploadErr +} +func (s *memoryStorage) Open(_ 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(_ context.Context, key string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.deleteErr != nil { + return s.deleteErr + } + delete(s.objects, key) + return nil +} +func (s *memoryStorage) count() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.objects) } + +type fixture struct { + pool *pgxpool.Pool + repo *game_imports_repo.GameImportsRepo + codec *game_package_service.GamePackageService + storage *memoryStorage + service *imports.GameImportService + actor int +} + +// newFixture creates isolated schema infrastructure; test data uses public repositories. +func newFixture(t *testing.T) fixture { + t.Helper() + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + t.Skip("set TEST_DATABASE_URL for atomic import tests") + } + ctx := context.Background() + admin, err := pgxpool.New(ctx, dsn) + check(t, err) + schema := fmt.Sprintf("import_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) + files, err := filepath.Glob("../../../migrations/*.sql") + check(t, err) + for _, filename := range files { + raw, err := os.ReadFile(filename) + check(t, err) + _, err = pool.Exec(ctx, strings.Split(string(raw), "-- +goose Down")[0]) + if err != nil { + t.Fatalf("migration %s: %v", filename, err) + } + } + actor, err := users_repo.NewUserRepo(pool).AddUserWithAgreements(ctx, "venue", "venue@test.invalid", "hash", []string{"organizer"}, + []user_agreements_repo.Agreement{{AgreementType: "terms", Version: "test"}}) + check(t, err) + f := fixture{ + pool: pool, + repo: game_imports_repo.NewGameImportsRepo(pool), + codec: game_package_service.NewGamePackageService(""), + storage: &memoryStorage{objects: map[string][]byte{}}, + actor: actor, + } + f.service = imports.NewGameImportService(f.repo, f.codec, f.storage, true) + return f +} + +func check(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } +} + +func snapshot() *game_package_service.Snapshot { + now := time.Date(2026, 10, 11, 5, 0, 0, 0, time.UTC) + owner, first, second := 1, 1, 2 + hash, used := sha256.Sum256([]byte("active-ticket")), sha256.Sum256([]byte("used-ticket")) + return &game_package_service.Snapshot{ + SourceServer: "https://online.example", + Data: game_package_service.Data{ + Game: game_package_service.Game{ID: 100, Name: "Offline game", StartAt: now, Status: "draft", Format: "offline", ScenarioID: 1, PlaylistID: &second}, + Scenario: game_package_service.Scenario{ID: 1, Name: "Source story", AuthorID: &owner, Status: "draft", CreatedAt: now, UpdatedAt: now, PlaylistID: &first, + Image: "image.png", Story: &storytelling.Story{ + Background: "image.png", + Introduction: &storytelling.Introduction{Text: "Intro", Audio: "music.mp3"}, + Conclusion: &storytelling.Conclusion{Text: "End", Video: "video.webm"}, + Places: []*storytelling.Place{{Code: "p", Name: "Room", Image: "image.png", Applications: []*storytelling.Application{{Name: "Clue", Image: "clue.pdf", FileType: "pdf"}}}}, + Questions: []*storytelling.Question{{Code: "q", Text: "Who?", Answer: "Secret"}}, + }}, + Teams: []game_package_service.Team{ + {ID: 200, GameID: 100, Name: "Confirmed", Status: "register", Password: "printed-qr", CreatorID: &owner, CaptainEmail: "captain@example.org", ParticipantsCount: 4, ParticipationStatus: "confirmed", ParticipationSource: "captain", CreatedAt: now}, + {ID: 201, GameID: 100, Name: "Cancelled", Status: "register", Password: "cancelled-qr", ParticipationStatus: "cancelled", CreatedAt: now}, + }, + Tickets: []game_package_service.Ticket{ + {ID: 300, GameID: 100, TokenHash: hash[:], Status: "active", CreatedAt: now}, + {ID: 301, GameID: 100, TokenHash: used[:], Status: "used", UsedAt: &now, UsedBy: &owner, CreatedAt: now}, + }, + Playlists: []game_package_service.Playlist{ + {ID: 1, Name: "Default", Tracks: []game_package_service.Track{{ID: 1, Name: "Same audio", Filename: "music.mp3", Position: 0}}}, + {ID: 2, Name: "Game music", Tracks: []game_package_service.Track{{ID: 2, Name: "Same audio again", Filename: "music.mp3", Position: 0}}}, + }, + ScreenPlan: &game_package_service.ScreenPlan{GameID: 100, CurrentStep: 1, Steps: []game_screen_plans_repo.Step{{Mode: 1}, {Mode: 8, Text: "Welcome"}}}, + Map: []game_package_service.MapItem{{GameID: 100, Kind: "team", ItemID: 200, X: 50, Y: 60, Hidden: true}, {GameID: 100, Kind: "organizer_desk", ItemID: 1, X: 80, Y: 90}}, + }, + } +} + +func media() map[string][]byte { + return map[string][]byte{ + "image.png": []byte("\x89PNG\r\n\x1a\nimage"), + "music.mp3": []byte("ID3\x04\x00\x00music"), + "video.webm": []byte("\x1a\x45\xdf\xa3video"), + "clue.pdf": []byte("%PDF-1.7\nclue"), + } +} + +func archive(t *testing.T, codec *game_package_service.GamePackageService, snapshot *game_package_service.Snapshot) []byte { + t.Helper() + files := media() + var out bytes.Buffer + check(t, codec.Pack(context.Background(), snapshot, func(_ context.Context, name string) (io.ReadCloser, error) { + body, ok := files[name] + if !ok { + return nil, os.ErrNotExist + } + return io.NopCloser(bytes.NewReader(body)), nil + }, &out)) + return out.Bytes() +} + +func TestImportPreservesGameAndIsIdempotent(t *testing.T) { + for _, tc := range []struct{ name, format string }{{"offline", "offline"}, {"online format retained", "online"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + ctx := context.Background() + scenarios := scenarios_repo.NewScenariosRepo(f.pool) + existing, err := scenarios.AddScenario(ctx, "Unrelated local story", f.actor) + check(t, err) + playlistRepo := playlists_repo.New(f.pool) + for _, name := range []string{"Local first", "Local second"} { + playlist, err := playlistRepo.Create(ctx, name) + check(t, err) + check(t, playlistRepo.AddTrack(ctx, playlist.ID, name, name+".mp3")) + } + source := snapshot() + source.Data.Game.Format = tc.format + raw := archive(t, f.codec, source) + result, err := f.service.Import(ctx, bytes.NewReader(raw), f.actor) + check(t, err) + if result.GameID != 100 || result.Repeated { + t.Fatal("incorrect result") + } + games := games_repo.NewGamesRepo(f.pool) + game, err := games.GetGameByID(ctx, 100) + check(t, err) + if game.Status != "draft" || game.Format != tc.format || game.ScenarioID == existing { + t.Fatal("game or scenario identity changed") + } + scenario, err := scenarios.GetScenarioByID(ctx, game.ScenarioID) + check(t, err) + var story storytelling.Story + check(t, json.Unmarshal([]byte(scenario.Scenario), &story)) + if scenario.Author == nil || scenario.Author.ID != f.actor || story.Questions[0].Answer != "Secret" { + t.Fatal("scenario ownership or answers lost") + } + old, err := scenarios.GetScenarioByID(ctx, existing) + check(t, err) + if old.Name != "Unrelated local story" { + t.Fatal("local scenario overwritten") + } + teams := teams_repo.NewTeamsRepo(f.pool) + team, err := teams.GetTeamByID(ctx, 200) + check(t, err) + if team.Password != "printed-qr" || team.CaptainEmail != "captain@example.org" || team.ParticipationStatus != "confirmed" { + t.Fatalf("team changed: %+v", team) + } + ticketRepo := tickets_repo.New(f.pool) + for _, ticket := range source.Data.Tickets { + stored, err := ticketRepo.GetByHash(ctx, ticket.TokenHash) + check(t, err) + if stored.ID != ticket.ID || stored.Status != ticket.Status || stored.UsedBy != nil || (stored.UsedAt == nil) != (ticket.UsedAt == nil) || (stored.UsedAt != nil && !stored.UsedAt.Equal(*ticket.UsedAt)) { + t.Fatal("ticket changed") + } + } + plan, err := game_screen_plans_repo.NewGameScreenPlansRepo(f.pool).Get(ctx, 100) + check(t, err) + if plan.CurrentStep != 1 || plan.Steps[1].Text != "Welcome" { + t.Fatal("screen plan lost") + } + items, err := game_map_repo.NewGameMapRepo(f.pool).List(ctx, 100) + check(t, err) + if len(items) != 2 || items[1].ItemID != 200 || !items[1].Hidden { + t.Fatal("map lost") + } + music, err := playlists_repo.New(f.pool).GetMusic(ctx, 100) + check(t, err) + if music.Playlist == nil || music.Playlist.Name != "Game music" || music.Playlist.ID == *source.Data.Game.PlaylistID { + t.Fatal("selected playlist lost or not copied") + } + // Every media object remains readable after the input and temporary files are gone. + for _, name := range []string{*scenario.Image, story.Background, story.Introduction.Audio, + story.Conclusion.Video, story.Places[0].Applications[0].Image, music.Playlist.Tracks[0].Filename} { + reader, err := f.storage.Open(ctx, name) + check(t, err) + body, err := io.ReadAll(reader) + check(t, err) + check(t, reader.Close()) + if len(body) == 0 { + t.Fatal("empty local media") + } + } + metadata, err := f.repo.Get(ctx, result.PackageID) + check(t, err) + var original game_package_service.Data + check(t, json.Unmarshal(metadata.OriginalData, &original)) + if original.Scenario.AuthorID == nil || *original.Scenario.AuthorID != 1 || original.Teams[0].CreatorID == nil || original.Tickets[1].UsedBy == nil || + metadata.Mappings.ScenarioID != game.ScenarioID || metadata.Mappings.Playlists[2] != music.Playlist.ID || len(metadata.Mappings.Tracks) != 2 { + t.Fatal("source data or ID mappings lost") + } + count := f.storage.count() + check(t, teams.UpdateTeam(ctx, 200, "Local edit")) + repeated, err := f.service.Import(ctx, bytes.NewReader(raw), f.actor) + check(t, err) + team, err = teams.GetTeamByID(ctx, 200) + check(t, err) + if !repeated.Repeated || *repeated != (imports.Result{100, result.PackageID, true}) || team.Name != "Local edit" || f.storage.count() != count { + t.Fatal("repeat changed local data") + } + nextGame, err := games.AddGame(ctx, "New local game", "", nil, time.Now(), int32(existing), "offline") + check(t, err) + nextTeam, err := teams.AddTeam(ctx, f.actor, nextGame, "Local team", "local-qr") + check(t, err) + hash := sha256.Sum256([]byte("new local ticket")) + nextTickets, err := ticketRepo.CreateBatch(ctx, nextGame, [][]byte{hash[:]}) + check(t, err) + if nextGame <= 100 || nextTeam <= 201 || nextTickets[0] <= 301 { + t.Fatal("sequence not advanced") + } + gameplay := gameplay_service.NewGameplayService(games, teams, scenarios_service.NewScenarioService(scenarios, cleaner.NewCleaner(), "https://online.example", nil, nil), storytelling.NewStory(cleaner.NewCleaner()), actions_repo.NewActionsRepo(f.pool), applications_repo.NewApplicationsRepo(f.pool), cleaner.NewCleaner(), "https://online.example") + _, _, _, err = gameplay.GetTeamActions(ctx, 200, "printed-qr") + check(t, err) + for _, access := range []struct { + id int + password string + }{{200, "wrong-password"}, {201, "cancelled-qr"}} { + _, _, _, err := gameplay.GetTeamActions(ctx, access.id, access.password) + if !errors.Is(err, gameplay_service.ErrTeamNotFound) { + t.Fatal("invalid or cancelled QR accepted") + } + } + }) + } +} + +func TestImportConflictsDoNotChangeData(t *testing.T) { + for _, tc := range []struct { + name string + change func(*game_package_service.Snapshot) + }{ + {"new package same game", func(s *game_package_service.Snapshot) {}}, + {"team ID collision", func(s *game_package_service.Snapshot) { + s.Data.Game.ID++ + for i := range s.Data.Teams { + s.Data.Teams[i].GameID++ + } + for i := range s.Data.Tickets { + s.Data.Tickets[i].GameID++ + } + s.Data.ScreenPlan.GameID++ + for i := range s.Data.Map { + s.Data.Map[i].GameID++ + } + }}, + {"ticket ID collision", func(s *game_package_service.Snapshot) { + s.Data.Game.ID++ + s.Data.Teams = nil + s.Data.Map = nil + s.Data.ScreenPlan = nil + for i := range s.Data.Tickets { + s.Data.Tickets[i].GameID++ + } + }}, + {"ticket hash collision", func(s *game_package_service.Snapshot) { + s.Data.Game.ID++ + s.Data.Teams = nil + s.Data.Map = nil + s.Data.ScreenPlan = nil + for i := range s.Data.Tickets { + s.Data.Tickets[i].GameID++ + s.Data.Tickets[i].ID += 10 + } + }}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + source := snapshot() + raw := archive(t, f.codec, source) + _, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + count := f.storage.count() + tc.change(source) + _, err = f.service.Import(context.Background(), bytes.NewReader(archive(t, f.codec, source)), f.actor) + if !errors.Is(err, imports.ErrConflict) || f.storage.count() != count { + t.Fatalf("conflict changed media or wrong error: %v", err) + } + team, err := teams_repo.NewTeamsRepo(f.pool).GetTeamByID(context.Background(), 200) + check(t, err) + if team.Password != "printed-qr" { + t.Fatal("conflict overwrote QR") + } + }) + } +} + +func TestImportFailureRecoversAndRetries(t *testing.T) { + for _, tc := range []struct { + name string + uploadFails, deleteFails bool + }{ + {"upload error", true, false}, {"cleanup retry after restart", true, true}, {"cancelled upload", false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + if tc.uploadFails { + f.storage.uploadErr = errors.New("partial upload") + } + if tc.deleteFails { + f.storage.deleteErr = errors.New("storage unavailable") + } + raw := archive(t, f.codec, snapshot()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if !tc.uploadFails { + f.storage.started, f.storage.release = make(chan struct{}, 1), make(chan struct{}) + go func() { <-f.storage.started; cancel() }() + } + _, err := f.service.Import(ctx, bytes.NewReader(raw), f.actor) + if err == nil { + t.Fatal("failure ignored") + } + _, err = games_repo.NewGamesRepo(f.pool).GetGameByID(context.Background(), 100) + if !errors.Is(err, games_repo.ErrGameNotFound) { + t.Fatal("partial game published") + } + f.storage.uploadErr, f.storage.deleteErr, f.storage.started = nil, nil, nil + restarted := imports.NewGameImportService(f.repo, f.codec, f.storage, true) + check(t, restarted.Maintain(context.Background())) + if f.storage.count() != 0 { + t.Fatal("recovery left objects") + } + result, err := restarted.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + if result.Repeated { + t.Fatal("failed attempt reused") + } + }) + } +} + +func TestConcurrentImportPublishesOnce(t *testing.T) { + for _, tc := range []struct{ name string }{{"same package"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + raw := archive(t, f.codec, snapshot()) + f.storage.started, f.storage.release = make(chan struct{}, 1), make(chan struct{}) + done := make(chan error, 1) + go func() { _, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor); done <- err }() + <-f.storage.started + _, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + if !errors.Is(err, imports.ErrInProgress) { + t.Fatalf("concurrent import: %v", err) + } + close(f.storage.release) + check(t, <-done) + check(t, f.service.Maintain(context.Background())) + if f.storage.count() != 5 { + t.Fatalf("unexpected objects: %d", f.storage.count()) + } + }) + } +} + +type ambiguousRepo struct { + *game_imports_repo.GameImportsRepo +} + +func (r ambiguousRepo) Complete(ctx context.Context, id string, raw []byte) error { + if err := r.GameImportsRepo.Complete(ctx, id, raw); err != nil { + return err + } + return errors.New("connection lost after commit") +} + +func TestAmbiguousCommitKeepsPublishedMedia(t *testing.T) { + for _, tc := range []struct{ name string }{{"committed but reply lost"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + raw := archive(t, f.codec, snapshot()) + service := imports.NewGameImportService(ambiguousRepo{f.repo}, f.codec, f.storage, true) + _, err := service.Import(context.Background(), bytes.NewReader(raw), f.actor) + if err == nil { + t.Fatal("ambiguous commit ignored") + } + check(t, f.service.Maintain(context.Background())) + if f.storage.count() != 5 { + t.Fatal("committed media deleted") + } + result, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + if !result.Repeated { + t.Fatal("committed package not recognized") + } + }) + } +} + +func TestImportRejectsInvalidInputBeforeStorage(t *testing.T) { + for _, tc := range []struct { + name string + venue bool + input []byte + cancel bool + want error + }{ + {"online server", false, nil, false, imports.ErrOnlineMode}, + {"broken ZIP", true, []byte("not a ZIP"), false, game_package_service.ErrInvalidPackage}, + {"cancelled receive", true, nil, true, context.Canceled}, + } { + t.Run(tc.name, func(t *testing.T) { + service := imports.NewGameImportService(nil, game_package_service.NewGamePackageService(""), nil, tc.venue) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if tc.cancel { + cancel() + } + _, err := service.Import(ctx, bytes.NewReader(tc.input), 1) + if !errors.Is(err, tc.want) { + t.Fatalf("got %v, want %v", err, tc.want) + } + }) + } +} + +func TestSamePackageIDWithDifferentBytesIsRejected(t *testing.T) { + for _, tc := range []struct{ name string }{{"changed ZIP comment"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + raw := archive(t, f.codec, snapshot()) + _, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + changed := append([]byte(nil), raw...) + binary.LittleEndian.PutUint16(changed[len(changed)-2:], 1) + changed = append(changed, 'x') + _, err = f.service.Import(context.Background(), bytes.NewReader(changed), f.actor) + if !errors.Is(err, imports.ErrConflict) { + t.Fatalf("changed package accepted: %v", err) + } + if f.storage.count() != 5 { + t.Fatal("changed package uploaded media") + } + }) + } +} + +func TestDeletedGameIsNotRestoredByRepeat(t *testing.T) { + for _, tc := range []struct{ name string }{{"deleted game tombstone"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + source := snapshot() + source.Data.Teams, source.Data.Tickets, source.Data.Map, source.Data.ScreenPlan = nil, nil, nil, nil + raw := archive(t, f.codec, source) + _, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + games := games_repo.NewGamesRepo(f.pool) + check(t, games.DeleteGameByID(context.Background(), 100)) + result, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + if !result.Repeated { + t.Fatal("tombstone forgotten") + } + _, err = games.GetGameByID(context.Background(), 100) + if !errors.Is(err, games_repo.ErrGameNotFound) { + t.Fatal("repeat restored deleted game") + } + }) + } +} + +type invalidPublicationRepo struct { + *game_imports_repo.GameImportsRepo +} + +func (r invalidPublicationRepo) Complete(ctx context.Context, id string, raw []byte) error { + var data game_package_service.Data + if err := json.Unmarshal(raw, &data); err != nil { + return err + } + data.Teams[0].Name = strings.Repeat("x", 256) + changed, err := json.Marshal(data) + if err != nil { + return err + } + return r.GameImportsRepo.Complete(ctx, id, changed) +} + +func TestPublicationRollsBackAllRows(t *testing.T) { + for _, tc := range []struct{ name string }{{"team constraint after game and scenario inserts"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + service := imports.NewGameImportService(invalidPublicationRepo{f.repo}, f.codec, f.storage, true) + raw := archive(t, f.codec, snapshot()) + _, err := service.Import(context.Background(), bytes.NewReader(raw), f.actor) + if !errors.Is(err, game_package_service.ErrInvalidPackage) { + t.Fatalf("wrong constraint error: %v", err) + } + _, err = games_repo.NewGamesRepo(f.pool).GetGameByID(context.Background(), 100) + if !errors.Is(err, games_repo.ErrGameNotFound) { + t.Fatal("partial game visible") + } + scenarios, err := scenarios_repo.NewScenariosRepo(f.pool).GetAllScenarios(context.Background()) + check(t, err) + playlists, err := playlists_repo.New(f.pool).List(context.Background()) + check(t, err) + if len(scenarios) != 0 || len(playlists) != 0 || f.storage.count() != 0 { + t.Fatal("rollback left rows or media") + } + }) + } +} + +func TestPublicationRechecksIDsAfterMediaUpload(t *testing.T) { + for _, tc := range []struct{ name string }{{"local game created while uploading"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + source := snapshot() + source.Data.Game.ID = 1 + for i := range source.Data.Teams { + source.Data.Teams[i].GameID = 1 + } + for i := range source.Data.Tickets { + source.Data.Tickets[i].GameID = 1 + } + for i := range source.Data.Map { + source.Data.Map[i].GameID = 1 + } + source.Data.ScreenPlan.GameID = 1 + raw := archive(t, f.codec, source) + f.storage.started, f.storage.release = make(chan struct{}, 1), make(chan struct{}) + done := make(chan error, 1) + go func() { _, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor); done <- err }() + <-f.storage.started + scenarioID, err := scenarios_repo.NewScenariosRepo(f.pool).AddScenario(context.Background(), "Local", f.actor) + check(t, err) + games := games_repo.NewGamesRepo(f.pool) + gameID, err := games.AddGame(context.Background(), "Local game", "", nil, time.Now(), int32(scenarioID), "offline") + check(t, err) + close(f.storage.release) + if err := <-done; !errors.Is(err, imports.ErrConflict) { + t.Fatalf("late conflict: %v", err) + } + game, err := games.GetGameByID(context.Background(), gameID) + check(t, err) + if game.Name != "Local game" || f.storage.count() != 0 { + t.Fatal("late conflict overwrote data or left media") + } + }) + } +} + +func TestImportWithoutOptionalData(t *testing.T) { + for _, tc := range []struct{ name string }{{"empty game without media or playlists"}} { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + source := snapshot() + source.Data.Game.PlaylistID = nil + source.Data.Scenario.PlaylistID = nil + source.Data.Scenario.Image = "" + source.Data.Scenario.Story = &storytelling.Story{} + source.Data.Teams, source.Data.Tickets, source.Data.Playlists, source.Data.Map, source.Data.ScreenPlan = nil, nil, nil, nil, nil + raw := archive(t, f.codec, source) + result, err := f.service.Import(context.Background(), bytes.NewReader(raw), f.actor) + check(t, err) + if result.GameID != 100 || f.storage.count() != 0 { + t.Fatal("empty game import failed") + } + }) + } +} + +func TestRecoveryRemovesOnlyOldImportDirectories(t *testing.T) { + for _, tc := range []struct { + name string + old, valid bool + removed bool + }{ + {"old unregistered request", true, true, true}, + {"request still being received", false, true, false}, + {"unrelated directory", true, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + id, err := file_storage.NewName() + check(t, err) + if !tc.valid { + id = "unrelated" + } + dir, err := os.MkdirTemp("", "game-import-"+id+"-") + check(t, err) + defer func() { _ = os.RemoveAll(dir) }() + if tc.old { + past := time.Now().Add(-33 * time.Minute) + check(t, os.Chtimes(dir, past, past)) + } + check(t, f.service.Maintain(context.Background())) + _, err = os.Stat(dir) + if errors.Is(err, os.ErrNotExist) != tc.removed { + t.Fatalf("wrong temporary cleanup: %v", err) + } + }) + } +} diff --git a/internal/services/game_import_service/media.go b/internal/services/game_import_service/media.go new file mode 100644 index 0000000..c4eb2cd --- /dev/null +++ b/internal/services/game_import_service/media.go @@ -0,0 +1,66 @@ +package game_import_service + +import ( + "path/filepath" + "sort" + "strconv" + + "evening_detective_server/internal/services/game_package_service" +) + +type mediaUpload struct { + key string + path string + mime string +} + +// prepareMedia changes only the validated bundle, whose original data has already been saved. +func prepareMedia(id string, bundle *game_package_service.Bundle) ([]mediaUpload, error) { + names := map[string]string{} + media := map[string]game_package_service.FileInfo{} + var uploads []mediaUpload + for _, file := range bundle.Manifest.Files { + if file.Path == "game.json" { + continue + } + key := "game-import-" + id + "-" + filepath.Base(file.Path) + names[file.Path] = key + media[file.Path] = file + uploads = append(uploads, mediaUpload{ + key: key, + path: bundle.Files[file.Path], + mime: file.MIME, + }) + } + // playlist_tracks.filename is unique even when two tracks use identical bytes. + trackNames := map[int]string{} + used := map[string]bool{} + for _, playlist := range bundle.Data.Playlists { + for _, track := range playlist.Tracks { + if !used[track.Filename] { + used[track.Filename] = true + trackNames[track.ID] = names[track.Filename] + continue + } + file := media[track.Filename] + key := "game-import-" + id + "-track-" + strconv.Itoa(track.ID) + filepath.Ext(track.Filename) + trackNames[track.ID] = key + uploads = append(uploads, mediaUpload{ + key: key, + path: bundle.Files[track.Filename], + mime: file.MIME, + }) + } + } + if err := game_package_service.RewriteMedia(&bundle.Data, names); err != nil { + return nil, err + } + for i := range bundle.Data.Playlists { + for j := range bundle.Data.Playlists[i].Tracks { + track := &bundle.Data.Playlists[i].Tracks[j] + track.Filename = trackNames[track.ID] + } + } + sort.Slice(uploads, func(i, j int) bool { return uploads[i].key < uploads[j].key }) + return uploads, nil +} diff --git a/internal/services/game_import_service/rustfs_test.go b/internal/services/game_import_service/rustfs_test.go new file mode 100644 index 0000000..81d3a3b --- /dev/null +++ b/internal/services/game_import_service/rustfs_test.go @@ -0,0 +1,137 @@ +package game_import_service_test + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + "evening_detective_server/internal/app" + "evening_detective_server/internal/modules/file_storage" + "evening_detective_server/internal/modules/processor_jwt" + imports "evening_detective_server/internal/services/game_import_service" + "github.com/kzzan/s3kit" +) + +func TestImportRustFSHTTPAndRestart(t *testing.T) { + for _, tc := range []struct{ name string }{{"venue HTTP, media, repeat and storage 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 smoke") + } + f := newFixture(t) + ctx := context.Background() + user, password := os.Getenv("TEST_S3_USER"), os.Getenv("TEST_S3_PASSWORD") + bucket := fmt.Sprintf("ed029-import-test-%d", time.Now().UnixNano()) + storage, err := file_storage.NewRustFSStorage(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 keys []string + t.Cleanup(func() { + for _, key := range keys { + if err := storage.Delete(ctx, key); err != nil { + t.Error(err) + } + } + if err := client.DeleteBucket(ctx, bucket); err != nil { + t.Error(err) + } + }) + service := imports.NewGameImportService(f.repo, f.codec, storage.(file_storage.IStreamStorage), true) + jwt := processor_jwt.NewProcessor("isolated-import-test-secret") + token, err := jwt.GenerateAccessToken(f.actor, "test@invalid", []string{"organizer"}) + check(t, err) + mux := http.NewServeMux() + mux.Handle("POST /api/games/import", app.GameImportHandler(jwt, service)) + server := httptest.NewServer(mux) + defer server.Close() + raw := archive(t, f.codec, snapshot()) + send := func() struct { + GameID int `json:"gameID"` + PackageID string `json:"packageID"` + Repeated bool `json:"repeated"` + } { + t.Helper() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, server.URL+"/api/games/import", bytes.NewReader(raw)) + check(t, err) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/zip") + response, err := server.Client().Do(req) + check(t, err) + defer func() { _ = response.Body.Close() }() + if response.StatusCode != 200 { + body, _ := io.ReadAll(response.Body) + t.Fatalf("import status %d: %s", response.StatusCode, body) + } + var result struct { + GameID int `json:"gameID"` + PackageID string `json:"packageID"` + Repeated bool `json:"repeated"` + } + check(t, json.NewDecoder(response.Body).Decode(&result)) + return result + } + first := send() + if first.GameID != 100 || first.Repeated { + t.Fatal("incorrect first import") + } + metadata, err := f.repo.Get(ctx, first.PackageID) + check(t, err) + keys = metadata.ObjectKeys + if len(keys) != 5 { + t.Fatal("lost media or duplicate track") + } + saved := map[string][]byte{} + for _, key := range keys { + file, err := storage.Get(ctx, key) + check(t, err) + if len(file.Data) == 0 { + t.Fatal("empty media") + } + saved[key] = file.Data + } + waitForImportStorageRestart(t) + reopened, err := file_storage.NewRustFSStorage(ctx, endpoint, user, password, bucket) + check(t, err) + for _, key := range keys { + file, err := reopened.Get(ctx, key) + check(t, err) + if !bytes.Equal(file.Data, saved[key]) { + t.Fatal("restart changed media") + } + } + repeated := send() + if !repeated.Repeated || repeated.PackageID != first.PackageID { + t.Fatal("repeat rebuilt game") + } + check(t, service.Maintain(ctx)) + }) + } +} + +// The external harness restarts only its isolated RustFS container. +func waitForImportStorageRestart(t *testing.T) { + t.Helper() + marker := os.Getenv("TEST_S3_RESTART_MARKER") + if marker == "" { + return + } + check(t, os.WriteFile(marker, []byte("ready"), 0600)) + defer func() { _ = os.Remove(marker); _ = os.Remove(marker + ".continue") }() + deadline := time.Now().Add(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 timed out") +} diff --git a/internal/services/game_import_service/service.go b/internal/services/game_import_service/service.go new file mode 100644 index 0000000..499026c --- /dev/null +++ b/internal/services/game_import_service/service.go @@ -0,0 +1,332 @@ +// Package game_import_service imports validated pre-game packages on the venue server. +package game_import_service + +import ( + "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/game_imports_repo" + "evening_detective_server/internal/services/game_package_service" +) + +var ErrOnlineMode = errors.New("импорт игры доступен только на площадке") + +// Result contains only the public import outcome. +type Result struct { + GameID int + PackageID string + Repeated bool +} + +// IGameImportsRepo is the persistence boundary for atomic publication and recovery. +type IGameImportsRepo interface { + Start(context.Context, *game_imports_repo.Attempt) (*game_imports_repo.Attempt, error) + Heartbeat(context.Context, string) error + Complete(context.Context, string, []byte) error + Fail(context.Context, string) error + Expire(context.Context) error + CleanupAttempts(context.Context) ([]game_imports_repo.Attempt, error) + MarkCleaned(context.Context, string) error + IsInactive(context.Context, string) (bool, error) +} + +type GameImportService struct { + repo IGameImportsRepo + codec *game_package_service.GamePackageService + storage file_storage.IStreamStorage + venue bool +} + +func NewGameImportService(repo IGameImportsRepo, codec *game_package_service.GamePackageService, + storage file_storage.IStreamStorage, venue bool) *GameImportService { + return &GameImportService{ + repo: repo, + codec: codec, + storage: storage, + venue: venue, + } +} + +// Import streams the request to disk; a repeated package never changes local data. +func (s *GameImportService) Import(ctx context.Context, input io.Reader, userID int) (result *Result, err error) { + defer func() { err = importError(err) }() + if !s.venue { + return nil, ErrOnlineMode + } + ctx, cancel := context.WithTimeout(ctx, game_imports_repo.PrepareTimeout) + defer cancel() + id, err := file_storage.NewName() + if err != nil { + return nil, err + } + dir, err := os.MkdirTemp("", "game-import-"+id+"-") + if err != nil { + return nil, err + } + defer func() { _ = os.RemoveAll(dir) }() + archive, err := os.CreateTemp(dir, "archive-*.zip") + if err != nil { + return nil, err + } + defer func() { _ = archive.Close() }() + digest := sha256.New() + size, err := copyContext(ctx, io.MultiWriter(archive, digest), io.LimitReader(input, game_package_service.MaxArchiveSize+1)) + if err != nil { + return nil, err + } + if size > game_package_service.MaxArchiveSize { + return nil, game_package_service.ErrLimitExceeded + } + bundle, err := s.codec.Unpack(ctx, archive, size, dir) + if err != nil { + return nil, err + } + if err := validateLocalIDs(&bundle.Data); err != nil { + return nil, err + } + original, err := json.Marshal(bundle.Data) + if err != nil { + return nil, err + } + manifest, err := json.Marshal(bundle.Manifest) + if err != nil { + return nil, err + } + uploads, err := prepareMedia(id, bundle) + if err != nil { + return nil, err + } + keys := make([]string, len(uploads)) + for i, upload := range uploads { + keys[i] = upload.key + } + attempt, err := s.repo.Start(ctx, &game_imports_repo.Attempt{ + ID: id, + PackageID: bundle.Manifest.PackageID, + SourceServer: bundle.Manifest.SourceServer, + GameID: bundle.Data.Game.ID, + SHA256: hex.EncodeToString(digest.Sum(nil)), + InitiatorID: userID, + Manifest: manifest, + OriginalData: original, + ObjectKeys: keys, + }) + if err != nil { + return nil, err + } + if attempt.State == "ready" { + return mapResult(attempt, true), nil + } + workCtx, stop := context.WithDeadline(ctx, attempt.CreatedAt.Add(game_imports_repo.PrepareTimeout)) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-workCtx.Done(): + return + case <-ticker.C: + if err := s.repo.Heartbeat(workCtx, id); err != nil { + stop() + return + } + } + } + }() + err = s.publish(workCtx, id, bundle, uploads) + stop() + <-done + if err == nil { + return mapResult(attempt, false), nil + } + recoveryCtx, recoveryCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer recoveryCancel() + failErr := s.repo.Fail(recoveryCtx, id) + // A committed or uncertain publication must keep its objects. + if failErr == nil { + failErr = s.cleanAttempt(recoveryCtx, game_imports_repo.Attempt{ + ID: id, + ObjectKeys: keys, + }) + } else if errors.Is(failErr, game_imports_repo.ErrOwnershipLost) { + failErr = nil + } + return nil, errors.Join(err, failErr) +} + +func (s *GameImportService) publish(ctx context.Context, id string, bundle *game_package_service.Bundle, uploads []mediaUpload) error { + for _, upload := range uploads { + if err := s.repo.Heartbeat(ctx, id); err != nil { + return err + } + file, err := os.Open(upload.path) + if err != nil { + return err + } + err = s.storage.PutStream(ctx, upload.key, file, upload.mime) + closeErr := file.Close() + if err := errors.Join(err, closeErr); err != nil { + return err + } + } + raw, err := json.Marshal(bundle.Data) + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + return s.repo.Complete(ctx, id, raw) +} + +func copyContext(ctx context.Context, output io.Writer, input io.Reader) (int64, error) { + var total int64 + buffer := make([]byte, 32<<10) + for { + if err := ctx.Err(); err != nil { + return total, err + } + n, readErr := input.Read(buffer) + if n > 0 { + written, err := output.Write(buffer[:n]) + total += int64(written) + if err != nil { + return total, err + } + if written != n { + return total, io.ErrShortWrite + } + } + if readErr == io.EOF { + return total, ctx.Err() + } + if readErr != nil { + return total, readErr + } + } +} + +// Run retries recovery after restart without an in-memory job queue. +func (s *GameImportService) Run(ctx context.Context) { + if !s.venue { + return + } + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + if err := s.Maintain(ctx); err != nil && ctx.Err() == nil { + log.Printf("game import recovery: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func (s *GameImportService) Maintain(ctx context.Context) error { + if !s.venue { + return ErrOnlineMode + } + if err := s.repo.Expire(ctx); err != nil { + return err + } + attempts, err := s.repo.CleanupAttempts(ctx) + if err != nil { + return err + } + for _, attempt := range attempts { + if err := s.cleanAttempt(ctx, attempt); err != nil { + return err + } + } + entries, err := os.ReadDir(os.TempDir()) + if err != nil { + return err + } + for _, entry := range entries { + if !entry.IsDir() || !strings.HasPrefix(entry.Name(), "game-import-") { + continue + } + suffix := strings.TrimPrefix(entry.Name(), "game-import-") + if len(suffix) < 34 || suffix[32] != '-' { + continue + } + if b, err := hex.DecodeString(suffix[:32]); err != nil || len(b) != 16 { + continue + } + info, err := entry.Info() + if err != nil { + return err + } + // Unregistered directories may still belong to a request being received or validated. + if time.Since(info.ModTime()) <= game_imports_repo.PrepareTimeout+game_imports_repo.Lease { + continue + } + inactive, err := s.repo.IsInactive(ctx, suffix[:32]) + if err != nil { + return err + } + if inactive { + if err := os.RemoveAll(filepath.Join(os.TempDir(), entry.Name())); err != nil { + return err + } + } + } + return ctx.Err() +} + +func (s *GameImportService) cleanAttempt(ctx context.Context, attempt game_imports_repo.Attempt) error { + for _, key := range attempt.ObjectKeys { + if err := s.storage.Delete(ctx, key); err != nil { + return err + } + } + return s.repo.MarkCleaned(ctx, attempt.ID) +} + +func validateLocalIDs(data *game_package_service.Data) error { + const maxID = 1<<31 - 2 + if data.Game.ID > maxID { + return fmt.Errorf("%w: game ID", game_package_service.ErrInvalidPackage) + } + for _, team := range data.Teams { + if team.ID > maxID { + return fmt.Errorf("%w: team ID", game_package_service.ErrInvalidPackage) + } + } + for _, ticket := range data.Tickets { + if ticket.ID > maxID { + return fmt.Errorf("%w: ticket ID", game_package_service.ErrInvalidPackage) + } + } + // Source IDs are cast to integer when allocating local copies. + if data.Scenario.ID > maxID { + return fmt.Errorf("%w: scenario ID", game_package_service.ErrInvalidPackage) + } + for _, playlist := range data.Playlists { + if playlist.ID > maxID { + return fmt.Errorf("%w: playlist ID", game_package_service.ErrInvalidPackage) + } + for _, track := range playlist.Tracks { + if track.ID > maxID { + return fmt.Errorf("%w: track ID", game_package_service.ErrInvalidPackage) + } + } + } + return nil +} diff --git a/internal/services/game_import_service/service_test.go b/internal/services/game_import_service/service_test.go new file mode 100644 index 0000000..a395830 --- /dev/null +++ b/internal/services/game_import_service/service_test.go @@ -0,0 +1,38 @@ +package game_import_service + +import ( + "context" + "errors" + "io" + "strings" + "testing" +) + +func TestCopyContextHonorsCancellationAndShortWrites(t *testing.T) { + for _, tc := range []struct { + name string + cancelled bool + output io.Writer + want error + }{ + {"cancelled", true, io.Discard, context.Canceled}, + {"short write", false, shortWriter{}, io.ErrShortWrite}, + {"normal", false, io.Discard, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if tc.cancelled { + cancel() + } + _, err := copyContext(ctx, tc.output, strings.NewReader("bytes")) + if !errors.Is(err, tc.want) { + t.Fatalf("got %v", err) + } + }) + } +} + +type shortWriter struct{} + +func (shortWriter) Write([]byte) (int, error) { return 0, nil } diff --git a/internal/services/game_package_service/service_test.go b/internal/services/game_package_service/service_test.go index 3ec74c0..9529be3 100644 --- a/internal/services/game_package_service/service_test.go +++ b/internal/services/game_package_service/service_test.go @@ -798,3 +798,38 @@ func TestRawSizeLimitCountsJSONAndMedia(t *testing.T) { }) } } + +func TestRewriteMediaChecksAllNamesBeforeMutation(t *testing.T) { + for _, tc := range []struct { + name string + complete bool + }{{"complete mapping", true}, {"missing mapping", false}} { + t.Run(tc.name, func(t *testing.T) { + snapshot, _ := fixture() + before, err := json.Marshal(snapshot.Data) + if err != nil { + t.Fatal(err) + } + names := map[string]string{"cover.png": "local.png", "music.mp3": "local.mp3", "video.webm": "local.webm", "evidence.pdf": "local.pdf"} + if !tc.complete { + delete(names, "evidence.pdf") + } + err = RewriteMedia(&snapshot.Data, names) + if !tc.complete { + after, _ := json.Marshal(snapshot.Data) + if !errors.Is(err, ErrInvalidPackage) || !bytes.Equal(before, after) { + t.Fatal("incomplete mapping changed data") + } + return + } + if err != nil { + t.Fatal(err) + } + for _, ref := range mediaRefs(&snapshot.Data) { + if *ref != "" && !strings.HasPrefix(*ref, "local.") { + t.Fatal("unmapped media") + } + } + }) + } +} diff --git a/internal/services/game_package_service/validation.go b/internal/services/game_package_service/validation.go index 4873033..0b9d902 100644 --- a/internal/services/game_package_service/validation.go +++ b/internal/services/game_package_service/validation.go @@ -195,3 +195,23 @@ func mediaRefs(data *Data) []*string { } return refs } + +// RewriteMedia replaces validated package media paths with local storage names. +// All names are checked before any reference is changed. +func RewriteMedia(data *Data, names map[string]string) error { + if err := validateData(data); err != nil { + return err + } + refs := mediaRefs(data) + for _, ref := range refs { + if *ref != "" && names[*ref] == "" { + return invalid("отсутствует локальное имя файла") + } + } + for _, ref := range refs { + if *ref != "" { + *ref = names[*ref] + } + } + return nil +}