generated from VLADIMIR/template
up
This commit is contained in:
@@ -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"
|
||||
}
|
||||
@@ -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, "Не удалось импортировать игру")
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 }
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user