diff --git a/cmd/api/handlers.go b/cmd/api/handlers.go index 066bca6..b73c3f7 100644 --- a/cmd/api/handlers.go +++ b/cmd/api/handlers.go @@ -16,7 +16,7 @@ func (app *application) ShowText(w http.ResponseWriter, r *http.Request) { return } - w.Write([]byte(fmt.Sprintf("Got world with name %s, core_version - %s, next_session - %v", world.ID, world.CoreVersion, world.NextSession))) + fmt.Fprintf(w, "Got world with name %s, core_version - %s, next_session - %v", world.ID, world.CoreVersion, world.NextSession) } func (app *application) WhenNextSession(w http.ResponseWriter, r *http.Request) { diff --git a/db/migrations/000002_create_game.up.sql b/db/migrations/000002_create_game.up.sql index 2be9a7e..0f92b04 100644 --- a/db/migrations/000002_create_game.up.sql +++ b/db/migrations/000002_create_game.up.sql @@ -268,7 +268,7 @@ CREATE TABLE IF NOT EXISTS ownership_string ( key_ VARCHAR(128) NOT NULL, value INTEGER NOT NULL, - card_deck_id TEXT UNIQUE, + card_deck_id TEXT, FOREIGN KEY (card_deck_id) REFERENCES card_deck(id) ON DELETE CASCADE ); diff --git a/internal/foundry/actions/session.go b/internal/foundry/actions/session.go index 2baf282..aadbab0 100644 --- a/internal/foundry/actions/session.go +++ b/internal/foundry/actions/session.go @@ -49,8 +49,7 @@ func (msg WsSessionMsg) OnActiveWorld(tr *transport.FoundryTransport) error { func (msg WsSessionMsg) HandleLoggedInUser(tr *transport.FoundryTransport) error { if tr.LoggedInChan == nil { tr.Logger.Info("World had been started before application was started. Run insertion of world data") - go tr.InsertGameToDB() - return nil + return tr.InsertGameToDB() } select { diff --git a/internal/foundry/foundry.go b/internal/foundry/foundry.go index 95caf76..40bc068 100644 --- a/internal/foundry/foundry.go +++ b/internal/foundry/foundry.go @@ -91,15 +91,23 @@ func (foundry *FoundryApi) ServeWebSocket() error { } go func() { - err = data.Action(foundry.Transport, &foundry.Status) - if err != nil { - wsChannels.Err() <- &types.FoundryError{Direction: types.ReaderCode, Err: err, Type: types.WebSocketCode} + errAction := data.Action(foundry.Transport, &foundry.Status) + + if errAction != nil { + wsChannels.Err() <- &types.FoundryError{Direction: types.ReaderCode, Err: errAction, Type: types.WebSocketCode} } }() case types.RespDataCode: - foundry.Transport.ExchangeChan.Msgs[message.Id] = make(chan []byte, 1) - foundry.Transport.ExchangeChan.Msgs[message.Id] <- []byte(message.MsgJson) - go foundry.Transport.CloseMsgChannel(message.Id, 5*time.Second) + tr := foundry.Transport + func() { + tr.ChanMutex.Lock() + defer tr.ChanMutex.Unlock() + + tr.ExchangeChan.Msgs[message.Id] = make(chan []byte, 1) + tr.ExchangeChan.Msgs[message.Id] <- []byte(message.MsgJson) + go tr.CloseMsgChannel(tr.ExchangeChan.Msgs[message.Id], message.Id, 5*time.Second) + + }() default: } case err = <-wsChannels.Err(): diff --git a/internal/foundry/models/db/actor.go b/internal/foundry/models/db/actor.go index 2132d25..74d5c90 100644 --- a/internal/foundry/models/db/actor.go +++ b/internal/foundry/models/db/actor.go @@ -34,11 +34,11 @@ func (a *Actor) Query(data *InsertId[uint]) { } func (a *Actor) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: a.ID, fieldName: "actor_id"} - InsertWithCtxParallel(group, ctx, tx, a.PrototypeToken, relId) - InsertWithCtxParallel(group, ctx, tx, a.Stats, relId) + InsertWithCtxParallel(group, tx, a.PrototypeToken, relId) + InsertWithCtxParallel(group, tx, a.Stats, relId) InsertSliceParallel(group, tx, a.Ownership, relId) InsertSliceParallel(group, tx, a.Items, relId) @@ -75,6 +75,10 @@ func (a *Actor) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[uint] args := []any{data.id, a.ID, a.Img, a.Name, a.Type, a.Folder, a.Sort} + mutex := GetMutex("actor_insert") + mutex.Lock() + defer mutex.Unlock() + var isInserted bool err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&isInserted) if err != nil { diff --git a/internal/foundry/models/db/authors.go b/internal/foundry/models/db/authors.go index 586b35c..239b766 100644 --- a/internal/foundry/models/db/authors.go +++ b/internal/foundry/models/db/authors.go @@ -46,6 +46,10 @@ func (a *Author) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[stri args := []any{data.id, a.Name, a.URL, a.Email, a.Discord} + mutex := GetMutex("author_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&a.ID) if err != nil { return err diff --git a/internal/foundry/models/db/card.go b/internal/foundry/models/db/card.go index a2a0e36..6362398 100644 --- a/internal/foundry/models/db/card.go +++ b/internal/foundry/models/db/card.go @@ -38,10 +38,10 @@ func (c *CardDeck) Query(data *InsertId[uint]) { } func (c *CardDeck) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: c.ID, fieldName: "card_deck_id"} - InsertWithCtxParallel(group, ctx, tx, c.Stats, relId) + InsertWithCtxParallel(group, tx, c.Stats, relId) InsertSliceParallel(group, tx, c.Ownership, relId) InsertSliceParallel(group, tx, c.Cards, relId) @@ -121,11 +121,11 @@ func (c *Card) Query(data *InsertId[string]) { } func (c *Card) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: c.ID, fieldName: "card_id"} - InsertWithCtxParallel(group, ctx, tx, c.Back, relId) - InsertWithCtxParallel(group, ctx, tx, c.Stats, relId) + InsertWithCtxParallel(group, tx, c.Back, relId) + InsertWithCtxParallel(group, tx, c.Stats, relId) InsertSliceParallel(group, tx, c.Faces, relId) err := group.Wait() diff --git a/internal/foundry/models/db/combat.go b/internal/foundry/models/db/combat.go index f083c6d..e21957c 100644 --- a/internal/foundry/models/db/combat.go +++ b/internal/foundry/models/db/combat.go @@ -34,10 +34,10 @@ func (c *Combat) Query(data *InsertId[uint]) { } func (c *Combat) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: c.ID, fieldName: "combat_id", tableName: "combat_groups"} - InsertWithCtxParallel(group, ctx, tx, c.Stats, relId) + InsertWithCtxParallel(group, tx, c.Stats, relId) InsertSimpleSliceParallel(group, tx, c.Groups, &relId) InsertSliceParallel(group, tx, c.Combatants, relId) @@ -113,9 +113,9 @@ func (c *Combatant) Query(data *InsertId[string]) { func (c *Combatant) InsertObjects(tx *sqlx.Tx) error { relId := InsertId[string]{id: c.ID, fieldName: "combatant_id"} - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, c.Stats, relId) + InsertWithCtxParallel(group, tx, c.Stats, relId) err := group.Wait() if err != nil && !errors.Is(err, sql.ErrNoRows) { diff --git a/internal/foundry/models/db/compatibility.go b/internal/foundry/models/db/compatibility.go index 5cc0532..182f106 100644 --- a/internal/foundry/models/db/compatibility.go +++ b/internal/foundry/models/db/compatibility.go @@ -44,6 +44,10 @@ func (c Compatibility) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertI args := []any{data.id, c.Minimum, c.Verified, c.Maximum} + mutex := GetMutex("compatibility_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&c.ID) if err != nil { return err diff --git a/internal/foundry/models/db/document_types.go b/internal/foundry/models/db/document_types.go index e494252..7f50542 100644 --- a/internal/foundry/models/db/document_types.go +++ b/internal/foundry/models/db/document_types.go @@ -52,6 +52,10 @@ func (d DocumentTypes) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertI args := []any{data.id} + mutex := GetMutex("document_types_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&d.ID) if err != nil { return err diff --git a/internal/foundry/models/db/folder.go b/internal/foundry/models/db/folder.go index 133de5b..2b1a298 100644 --- a/internal/foundry/models/db/folder.go +++ b/internal/foundry/models/db/folder.go @@ -100,10 +100,10 @@ func (w *WorldFolder) Query(data *InsertId[uint]) { } func (w *WorldFolder) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: w.ID, fieldName: "world_folder_id"} - InsertWithCtxParallel(group, ctx, tx, w.Stats, relId) + InsertWithCtxParallel(group, tx, w.Stats, relId) err := group.Wait() if err != nil && !errors.Is(err, sql.ErrNoRows) { diff --git a/internal/foundry/models/db/game.go b/internal/foundry/models/db/game.go index bac1ade..1ab4ced 100644 --- a/internal/foundry/models/db/game.go +++ b/internal/foundry/models/db/game.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "fmt" "strconv" "time" @@ -47,53 +48,81 @@ type Game struct { // Scenes []Scene } -func (g *Game) InsertObjects(tx *sqlx.Tx) error { +func (g *Game) InsertObjects(db *sqlx.DB) error { relData := InsertId[uint]{id: g.ID, fieldName: "game_id"} - group, ctx := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, &g.Addresses, relData) - InsertWithCtxParallel(group, ctx, tx, &g.Files, relData) - InsertWithCtxParallel(group, ctx, tx, &g.Options, relData) - InsertWithCtxParallel(group, ctx, tx, &g.Release, relData) - InsertWithCtxParallel(group, ctx, tx, &g.CoreUpdate, relData) - InsertWithCtxParallel(group, ctx, tx, &g.SystemUpdate, relData) + err := func() error { + tx := db.MustBegin() + defer tx.Rollback() - relDataString := InsertId[string]{id: strconv.FormatUint(uint64(g.ID), 10), fieldName: "game_id"} - InsertWithCtxParallel(group, ctx, tx, g.World, relDataString) - InsertWithCtxParallel(group, ctx, tx, g.System, relDataString) + groupFunc, _ := errgroup.WithContext(context.Background()) - InsertSimpleSliceParallel(group, tx, g.ActiveUsers, - &InsertId[uint]{id: g.ID, fieldName: "game_id", tableName: "active_users"}) + InsertWithCtxParallel(groupFunc, tx, &g.Addresses, relData) + InsertWithCtxParallel(groupFunc, tx, &g.Files, relData) + InsertWithCtxParallel(groupFunc, tx, &g.Options, relData) + InsertWithCtxParallel(groupFunc, tx, &g.Release, relData) + InsertWithCtxParallel(groupFunc, tx, &g.CoreUpdate, relData) + InsertWithCtxParallel(groupFunc, tx, &g.SystemUpdate, relData) - InsertSliceParallel(group, tx, g.Modules, relData) - InsertSliceParallel(group, tx, g.PackageWarnings, relData) - InsertSliceParallel(group, tx, g.Packs, relDataString) - InsertSliceParallel(group, tx, g.Messages, relData) - InsertSliceParallel(group, tx, g.Combats, relData) - InsertSliceParallel(group, tx, g.CardDeck, relData) - InsertSliceParallel(group, tx, g.Users, relData) - InsertSliceParallel(group, tx, g.Macros, relData) - InsertSliceParallel(group, tx, g.Folders, relData) - InsertSliceParallel(group, tx, g.Settings, relData) - InsertSliceParallel(group, tx, g.Journals, relData) - InsertSliceParallel(group, tx, g.Tables, relData) - InsertSliceParallel(group, tx, g.Playlists, relData) + relDataString := InsertId[string]{id: strconv.FormatUint(uint64(g.ID), 10), fieldName: "game_id"} + InsertWithCtxParallel(groupFunc, tx, g.World, relDataString) + InsertWithCtxParallel(groupFunc, tx, g.System, relDataString) - err := group.Wait() - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return err - } + InsertSimpleSliceParallel(groupFunc, tx, g.ActiveUsers, + &InsertId[uint]{id: g.ID, fieldName: "game_id", tableName: "active_users"}) - InsertSliceParallelTimeout(group, tx, g.Items, - InsertId[string]{id: strconv.FormatUint(uint64(g.ID), 10), fieldName: "game_id"}, - 15*time.Second) - InsertSliceParallelTimeout(group, tx, g.Actors, relData, 15*time.Second) + InsertSliceParallel(groupFunc, tx, g.Modules, relData) + InsertSliceParallel(groupFunc, tx, g.PackageWarnings, relData) + InsertSliceParallel(groupFunc, tx, g.Packs, relDataString) + InsertSliceParallel(groupFunc, tx, g.Messages, relData) + InsertSliceParallel(groupFunc, tx, g.Combats, relData) + InsertSliceParallel(groupFunc, tx, g.CardDeck, relData) + InsertSliceParallel(groupFunc, tx, g.Users, relData) + InsertSliceParallel(groupFunc, tx, g.Macros, relData) + InsertSliceParallel(groupFunc, tx, g.Folders, relData) - err = group.Wait() - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return err - } - return nil + errFunc := groupFunc.Wait() + if errFunc != nil && !errors.Is(errFunc, sql.ErrNoRows) { + return errFunc + } + return tx.Commit() + }() + + err = func() error { + tx := db.MustBegin() + defer tx.Rollback() + groupFunc, _ := errgroup.WithContext(context.Background()) + + InsertSliceParallel(groupFunc, tx, g.Settings, relData) + InsertSliceParallel(groupFunc, tx, g.Journals, relData) + InsertSliceParallel(groupFunc, tx, g.Tables, relData) + InsertSliceParallel(groupFunc, tx, g.Playlists, relData) + + errFunc := groupFunc.Wait() + if errFunc != nil && !errors.Is(errFunc, sql.ErrNoRows) { + return errFunc + } + return tx.Commit() + }() + + err = func() error { + tx := db.MustBegin() + defer tx.Rollback() + groupFunc, _ := errgroup.WithContext(context.Background()) + + InsertSliceParallelTimeout(groupFunc, tx, g.Items, + InsertId[string]{id: strconv.FormatUint(uint64(g.ID), 10), fieldName: "game_id"}, + 15*time.Second) + InsertSliceParallelTimeout(groupFunc, tx, g.Actors, relData, 15*time.Second) + + errFunc := groupFunc.Wait() + if errFunc != nil && !errors.Is(errFunc, sql.ErrNoRows) { + return errFunc + } + return tx.Commit() + }() + + return err } func (g *Game) Insert(db *sqlx.DB) error { @@ -114,11 +143,15 @@ func (g *Game) Insert(db *sqlx.DB) error { if err != nil { return err } - - err = g.InsertObjects(tx) + err = tx.Commit() if err != nil { return err } - return tx.Commit() + err = g.InsertObjects(db) + if err != nil { + fmt.Printf("%v\n", err) + } + + return err } diff --git a/internal/foundry/models/db/index.go b/internal/foundry/models/db/index.go index 1df0d4a..2d68235 100644 --- a/internal/foundry/models/db/index.go +++ b/internal/foundry/models/db/index.go @@ -76,6 +76,10 @@ func (i *Index) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[strin args := []any{i.ID, i.Folder, i.Img, i.Name, i.Type} + mutex := GetMutex("index_insert") + mutex.Lock() + defer mutex.Unlock() + _, err := tx.ExecContext(ctx, data.query, args...) if err != nil { return err diff --git a/internal/foundry/models/db/item.go b/internal/foundry/models/db/item.go index 17f520d..f238ef7 100644 --- a/internal/foundry/models/db/item.go +++ b/internal/foundry/models/db/item.go @@ -33,10 +33,10 @@ func (i *Item) Query(data *InsertId[string]) { } func (i *Item) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: i.ID, fieldName: "item_id"} - InsertWithCtxParallel(group, ctx, tx, i.Stats, relId) + InsertWithCtxParallel(group, tx, i.Stats, relId) InsertSliceParallel(group, tx, i.Ownership, relId) err := group.Wait() @@ -72,6 +72,10 @@ func (i *Item) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[string args := []any{data.id, i.ID, i.Img, i.Name, i.Type, i.Folder, i.Sort} + mutex := GetMutex("item_insert") + mutex.Lock() + defer mutex.Unlock() + var isInserted bool err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&isInserted) if err != nil { diff --git a/internal/foundry/models/db/journal_page.go b/internal/foundry/models/db/journal_page.go index 479de5f..1a87279 100644 --- a/internal/foundry/models/db/journal_page.go +++ b/internal/foundry/models/db/journal_page.go @@ -35,13 +35,13 @@ func (j *JournalPage) Query(data *InsertId[string]) { } func (j *JournalPage) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: j.ID, fieldName: "journal_page_id"} - InsertWithCtxParallel(group, ctx, tx, j.Text, relId) - InsertWithCtxParallel(group, ctx, tx, j.Title, relId) - InsertWithCtxParallel(group, ctx, tx, j.Video, relId) - InsertWithCtxParallel(group, ctx, tx, j.Stats, relId) + InsertWithCtxParallel(group, tx, j.Text, relId) + InsertWithCtxParallel(group, tx, j.Title, relId) + InsertWithCtxParallel(group, tx, j.Video, relId) + InsertWithCtxParallel(group, tx, j.Stats, relId) InsertSliceParallel(group, tx, j.Ownership, relId) diff --git a/internal/foundry/models/db/language.go b/internal/foundry/models/db/language.go index 47cb892..18c8fd3 100644 --- a/internal/foundry/models/db/language.go +++ b/internal/foundry/models/db/language.go @@ -145,6 +145,10 @@ func (l *Language) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[st args := []any{data.id, l.Lang, l.Name, l.Path} + mutex := GetMutex("language_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&l.ID) if err != nil { return err diff --git a/internal/foundry/models/db/light.go b/internal/foundry/models/db/light.go index bc95156..8420e0d 100644 --- a/internal/foundry/models/db/light.go +++ b/internal/foundry/models/db/light.go @@ -39,11 +39,11 @@ func (l *Light) Query(data *InsertId[uint]) { } func (l *Light) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[uint]{id: l.ID, fieldName: "token_light_id"} - InsertWithCtxParallel(group, ctx, tx, l.LightAnimation, relId) - InsertWithCtxParallel(group, ctx, tx, l.LightDarkness, relId) + InsertWithCtxParallel(group, tx, l.LightAnimation, relId) + InsertWithCtxParallel(group, tx, l.LightDarkness, relId) err := group.Wait() if err != nil && !errors.Is(err, sql.ErrNoRows) { diff --git a/internal/foundry/models/db/macro.go b/internal/foundry/models/db/macro.go index 501f513..7b0f1e0 100644 --- a/internal/foundry/models/db/macro.go +++ b/internal/foundry/models/db/macro.go @@ -35,10 +35,10 @@ func (m *Macro) Query(data *InsertId[uint]) { } func (m *Macro) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: m.ID, fieldName: "macro_id"} - InsertWithCtxParallel(group, ctx, tx, m.Stats, relId) + InsertWithCtxParallel(group, tx, m.Stats, relId) InsertSliceParallel(group, tx, m.Ownership, relId) err := group.Wait() diff --git a/internal/foundry/models/db/media.go b/internal/foundry/models/db/media.go index 32cb42f..98ed234 100644 --- a/internal/foundry/models/db/media.go +++ b/internal/foundry/models/db/media.go @@ -44,6 +44,10 @@ func (m *Media) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[strin args := []any{data.id, m.Type, m.URL, m.Caption} + mutex := GetMutex("media_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&m.ID) if err != nil { return err diff --git a/internal/foundry/models/db/message.go b/internal/foundry/models/db/message.go index c3df510..009f7fa 100644 --- a/internal/foundry/models/db/message.go +++ b/internal/foundry/models/db/message.go @@ -40,10 +40,10 @@ func (m *Message) Query(data *InsertId[uint]) { func (m *Message) InsertObjects(tx *sqlx.Tx) error { relId := InsertId[string]{id: m.ID, fieldName: "message_id"} - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, m.Stats, relId) - InsertWithCtxParallel(group, ctx, tx, m.Speaker, relId) + InsertWithCtxParallel(group, tx, m.Stats, relId) + InsertWithCtxParallel(group, tx, m.Speaker, relId) InsertSimpleSliceParallel(group, tx, m.Whisper, &InsertId[string]{id: m.ID, fieldName: "message_id", tableName: "message_whisper"}) InsertSimpleSliceParallel(group, tx, m.Rolls, &InsertId[string]{id: m.ID, fieldName: "message_id", tableName: "message_rolls"}) diff --git a/internal/foundry/models/db/module.go b/internal/foundry/models/db/module.go index b59bfef..debef9e 100644 --- a/internal/foundry/models/db/module.go +++ b/internal/foundry/models/db/module.go @@ -66,11 +66,11 @@ func (m *Module) ConnectGameQuery(data *InsertId[uint]) { func (m *Module) InsertObjects(tx *sqlx.Tx) error { relId := InsertId[string]{id: m.ID, fieldName: "module_id"} - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, m.DocumentTypes, relId) - InsertWithCtxParallel(group, ctx, tx, m.Relationships, relId) - InsertWithCtxParallel(group, ctx, tx, m.Compatibility, relId) + InsertWithCtxParallel(group, tx, m.DocumentTypes, relId) + InsertWithCtxParallel(group, tx, m.Relationships, relId) + InsertWithCtxParallel(group, tx, m.Compatibility, relId) scriptRelId := &InsertId[string]{id: m.ID, fieldName: "module_id", tableName: "scripts"} InsertSimpleSliceParallel(group, tx, m.Scripts, scriptRelId) diff --git a/internal/foundry/models/db/ownership.go b/internal/foundry/models/db/ownership.go index 51efcf5..e81e409 100644 --- a/internal/foundry/models/db/ownership.go +++ b/internal/foundry/models/db/ownership.go @@ -88,6 +88,10 @@ func (o OwnershipString) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *Inser args := []any{data.id, o.Key, o.Value} + mutex := GetMutex("ownership_string_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&o.ID) if err != nil { return err diff --git a/internal/foundry/models/db/pack.go b/internal/foundry/models/db/pack.go index 670face..2868f16 100644 --- a/internal/foundry/models/db/pack.go +++ b/internal/foundry/models/db/pack.go @@ -48,9 +48,9 @@ func (p *Pack) ConnectGameQuery(data *InsertId[string]) { } func (p *Pack) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, p.Ownership, + InsertWithCtxParallel(group, tx, p.Ownership, InsertId[string]{id: p.ID, fieldName: "pack_id"}) relId := InsertId[string]{id: p.ID, fieldName: "pack_id"} @@ -88,6 +88,10 @@ func (p *Pack) Insert(tx *sqlx.Tx, data *InsertId[string]) error { args := []any{dataId, p.ID, p.Name, p.Label, p.Banner, p.Path, p.Type, p.System, p.PackageType, p.PackageName} + mutex := GetMutex("pack_insert") + mutex.Lock() + defer mutex.Unlock() + res, err := tx.Exec(data.query, args...) if err != nil { return err @@ -134,6 +138,10 @@ func (p *Pack) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[string args := []any{dataId, p.ID, p.Name, p.Label, p.Banner, p.Path, p.Type, p.System, p.PackageType, p.PackageName} + mutex := GetMutex("pack_insert") + mutex.Lock() + defer mutex.Unlock() + res, err := tx.ExecContext(ctx, data.query, args...) if err != nil { return err diff --git a/internal/foundry/models/db/package_warnings.go b/internal/foundry/models/db/package_warnings.go index 12eee8e..c3de4f1 100644 --- a/internal/foundry/models/db/package_warnings.go +++ b/internal/foundry/models/db/package_warnings.go @@ -48,6 +48,10 @@ func (p *PackageWarning) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *Inser args := []any{data.id, p.Key} + mutex := GetMutex("package_warning_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&p.ID) if err != nil { return err diff --git a/internal/foundry/models/db/playlist.go b/internal/foundry/models/db/playlist.go index 9b69f09..35438d3 100644 --- a/internal/foundry/models/db/playlist.go +++ b/internal/foundry/models/db/playlist.go @@ -38,10 +38,10 @@ func (p *Playlist) Query(data *InsertId[uint]) { } func (p *Playlist) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: p.ID, fieldName: "playlist_id"} - InsertWithCtxParallel(group, ctx, tx, p.Stats, relId) + InsertWithCtxParallel(group, tx, p.Stats, relId) InsertSliceParallel(group, tx, p.Ownership, relId) InsertSliceParallel(group, tx, p.Sounds, relId) diff --git a/internal/foundry/models/db/relationships.go b/internal/foundry/models/db/relationships.go index 72e1e63..9858054 100644 --- a/internal/foundry/models/db/relationships.go +++ b/internal/foundry/models/db/relationships.go @@ -60,6 +60,10 @@ func (r Relationships) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertI args := []any{data.id} + mutex := GetMutex("relationships_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&r.ID) if err != nil { return err @@ -106,8 +110,8 @@ func (r RelationshipsData) Insert(tx *sqlx.Tx, data *InsertId[uint]) error { return err } - group, ctx := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, r.Compatibility, InsertId[string]{id: strconv.FormatUint(uint64(r.ID), 10), fieldName: fmt.Sprintf("%s_id", data.tableName)}) + group, _ := errgroup.WithContext(context.Background()) + InsertWithCtxParallel(group, tx, r.Compatibility, InsertId[string]{id: strconv.FormatUint(uint64(r.ID), 10), fieldName: fmt.Sprintf("%s_id", data.tableName)}) return group.Wait() } @@ -124,8 +128,8 @@ func (r RelationshipsData) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *Ins return err } - group, ctx := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, r.Compatibility, InsertId[string]{id: strconv.FormatUint(uint64(r.ID), 10), fieldName: fmt.Sprintf("%s_id", data.tableName)}) + group, _ := errgroup.WithContext(context.Background()) + InsertWithCtxParallel(group, tx, r.Compatibility, InsertId[string]{id: strconv.FormatUint(uint64(r.ID), 10), fieldName: fmt.Sprintf("%s_id", data.tableName)}) return group.Wait() } diff --git a/internal/foundry/models/db/ring.go b/internal/foundry/models/db/ring.go index 672968e..f65b0ad 100644 --- a/internal/foundry/models/db/ring.go +++ b/internal/foundry/models/db/ring.go @@ -27,11 +27,11 @@ func (r *Ring) Query(data *InsertId[uint]) { } func (r *Ring) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[uint]{id: r.ID, fieldName: "ring_id"} - InsertWithCtxParallel(group, ctx, tx, r.RingColors, relId) - InsertWithCtxParallel(group, ctx, tx, r.Subject, relId) + InsertWithCtxParallel(group, tx, r.RingColors, relId) + InsertWithCtxParallel(group, tx, r.Subject, relId) err := group.Wait() if err != nil && !errors.Is(err, sql.ErrNoRows) { diff --git a/internal/foundry/models/db/settings.go b/internal/foundry/models/db/settings.go index c480373..1c11a68 100644 --- a/internal/foundry/models/db/settings.go +++ b/internal/foundry/models/db/settings.go @@ -28,10 +28,10 @@ func (s *Setting) Query(data *InsertId[uint]) { } func (s *Setting) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: s.ID, fieldName: "setting_id"} - InsertWithCtxParallel(group, ctx, tx, s.Stats, relId) + InsertWithCtxParallel(group, tx, s.Stats, relId) err := group.Wait() if err != nil && !errors.Is(err, sql.ErrNoRows) { diff --git a/internal/foundry/models/db/setup.go b/internal/foundry/models/db/setup.go index 9ab41ce..371a3ad 100644 --- a/internal/foundry/models/db/setup.go +++ b/internal/foundry/models/db/setup.go @@ -32,13 +32,13 @@ type Setup struct { func (s *Setup) InsertObjects(tx *sqlx.Tx) error { relData := InsertId[uint]{id: s.ID, fieldName: "setup_id"} - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, &s.CoreUpdate, relData) - InsertWithCtxParallel(group, ctx, tx, &s.FeaturedContent, relData) - InsertWithCtxParallel(group, ctx, tx, &s.Files, relData) - InsertWithCtxParallel(group, ctx, tx, s.Options, relData) - InsertWithCtxParallel(group, ctx, tx, &s.Release, relData) + InsertWithCtxParallel(group, tx, &s.CoreUpdate, relData) + InsertWithCtxParallel(group, tx, &s.FeaturedContent, relData) + InsertWithCtxParallel(group, tx, &s.Files, relData) + InsertWithCtxParallel(group, tx, s.Options, relData) + InsertWithCtxParallel(group, tx, &s.Release, relData) InsertSliceParallel(group, tx, s.Languages, relData) InsertSliceParallel(group, tx, s.Modules, relData) diff --git a/internal/foundry/models/db/stats.go b/internal/foundry/models/db/stats.go index 90b4fcf..e300c7f 100644 --- a/internal/foundry/models/db/stats.go +++ b/internal/foundry/models/db/stats.go @@ -45,6 +45,10 @@ func (s Stats) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[string args := []any{data.id, s.CoreVersion, s.SystemID, s.SystemVersion, s.LastModifiedBy, s.ModifiedTime} + mutex := GetMutex("stats_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&s.ID) if err != nil { return err diff --git a/internal/foundry/models/db/style.go b/internal/foundry/models/db/style.go index f692d3e..e93db24 100644 --- a/internal/foundry/models/db/style.go +++ b/internal/foundry/models/db/style.go @@ -42,6 +42,10 @@ func (s *Style) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[strin args := []any{data.id, s.Src} + mutex := GetMutex("style_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&s.ID) if err != nil { return err diff --git a/internal/foundry/models/db/system.go b/internal/foundry/models/db/system.go index b19fe5c..2fb7242 100644 --- a/internal/foundry/models/db/system.go +++ b/internal/foundry/models/db/system.go @@ -64,12 +64,12 @@ func (s *System) ConnectGameQuery(data *InsertId[string]) { func (s *System) InsertObjects(tx *sqlx.Tx) error { relId := InsertId[string]{id: s.ID, fieldName: "system_id"} - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, s.Compatibility, relId) - InsertWithCtxParallel(group, ctx, tx, s.Relationships, relId) - InsertWithCtxParallel(group, ctx, tx, s.DocumentTypes, relId) - InsertWithCtxParallel(group, ctx, tx, s.Grid, relId) + InsertWithCtxParallel(group, tx, s.Compatibility, relId) + InsertWithCtxParallel(group, tx, s.Relationships, relId) + InsertWithCtxParallel(group, tx, s.DocumentTypes, relId) + InsertWithCtxParallel(group, tx, s.Grid, relId) esModulesRelId := &InsertId[string]{id: s.ID, fieldName: "system_id", tableName: "es_modules"} InsertSimpleSliceParallel(group, tx, s.Esmodules, esModulesRelId) diff --git a/internal/foundry/models/db/table.go b/internal/foundry/models/db/table.go index 2b12769..561128c 100644 --- a/internal/foundry/models/db/table.go +++ b/internal/foundry/models/db/table.go @@ -36,10 +36,10 @@ func (t *Table) Query(data *InsertId[uint]) { } func (t *Table) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: t.ID, fieldName: "table_id"} - InsertWithCtxParallel(group, ctx, tx, t.Stats, relId) + InsertWithCtxParallel(group, tx, t.Stats, relId) InsertSliceParallel(group, tx, t.Ownership, relId) InsertSliceParallel(group, tx, t.Results, relId) @@ -112,9 +112,9 @@ func (t *TableResult) Query(data *InsertId[string]) { } func (t *TableResult) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, t.Stats, InsertId[string]{id: t.ID, fieldName: "table_result_id"}) + InsertWithCtxParallel(group, tx, t.Stats, InsertId[string]{id: t.ID, fieldName: "table_result_id"}) InsertSimpleSlice(tx, t.Range, &InsertId[string]{id: t.ID, fieldName: "table_result_id", tableName: "table_result_range"}) err := group.Wait() diff --git a/internal/foundry/models/db/token.go b/internal/foundry/models/db/token.go index a0954a0..433d566 100644 --- a/internal/foundry/models/db/token.go +++ b/internal/foundry/models/db/token.go @@ -45,17 +45,17 @@ func (t *Token) Query(data *InsertId[string]) { } func (t *Token) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[uint]{id: t.ID, fieldName: "token_id"} - InsertWithCtxParallel(group, ctx, tx, t.Ring, relId) - InsertWithCtxParallel(group, ctx, tx, t.Sight, relId) - InsertWithCtxParallel(group, ctx, tx, t.Texture, relId) - InsertWithCtxParallel(group, ctx, tx, t.Bar1, InsertId[uint]{id: t.ID, fieldName: "token_id", tableName: "token_bar_1"}) - InsertWithCtxParallel(group, ctx, tx, t.Bar1, InsertId[uint]{id: t.ID, fieldName: "token_id", tableName: "token_bar_2"}) - InsertWithCtxParallel(group, ctx, tx, t.Light, relId) - InsertWithCtxParallel(group, ctx, tx, t.Occludable, relId) - InsertWithCtxParallel(group, ctx, tx, t.TurnMarker, relId) + InsertWithCtxParallel(group, tx, t.Ring, relId) + InsertWithCtxParallel(group, tx, t.Sight, relId) + InsertWithCtxParallel(group, tx, t.Texture, relId) + InsertWithCtxParallel(group, tx, t.Bar1, InsertId[uint]{id: t.ID, fieldName: "token_id", tableName: "token_bar_1"}) + InsertWithCtxParallel(group, tx, t.Bar1, InsertId[uint]{id: t.ID, fieldName: "token_id", tableName: "token_bar_2"}) + InsertWithCtxParallel(group, tx, t.Light, relId) + InsertWithCtxParallel(group, tx, t.Occludable, relId) + InsertWithCtxParallel(group, tx, t.TurnMarker, relId) err := group.Wait() if err != nil && !errors.Is(err, sql.ErrNoRows) { @@ -88,6 +88,10 @@ func (t *Token) InsertCtx(ctx context.Context, tx *sqlx.Tx, data *InsertId[strin args := []any{data.id, t.Name, t.ActorLink, t.AppendNumber, t.PrependAdjective, t.LockRotation, t.RandomImg, t.DisplayName, t.DisplayBars, t.Disposition, t.Rotation, t.Alpha, t.Width, t.Height} + mutex := GetMutex("token_insert") + mutex.Lock() + defer mutex.Unlock() + err := tx.QueryRowxContext(ctx, data.query, args...).Scan(&t.ID) if err != nil { return err diff --git a/internal/foundry/models/db/user.go b/internal/foundry/models/db/user.go index 1e553d9..ccfb2cf 100644 --- a/internal/foundry/models/db/user.go +++ b/internal/foundry/models/db/user.go @@ -33,10 +33,10 @@ func (u *User) Query(data *InsertId[uint]) { } func (u *User) InsertObjects(tx *sqlx.Tx) error { - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) relId := InsertId[string]{id: u.ID, fieldName: "user_id"} - InsertWithCtxParallel(group, ctx, tx, u.Stats, relId) + InsertWithCtxParallel(group, tx, u.Stats, relId) InsertSliceParallel(group, tx, u.Hotbar, relId) err := group.Wait() diff --git a/internal/foundry/models/db/utils.go b/internal/foundry/models/db/utils.go index fdbb73c..f342f32 100644 --- a/internal/foundry/models/db/utils.go +++ b/internal/foundry/models/db/utils.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "time" "github.com/jmoiron/sqlx" @@ -41,7 +42,7 @@ func InsertWithCtx[T AllowedIds, I Insertable[T]](tx *sqlx.Tx, data I, relId Ins return data.InsertCtx(ctx, tx, &relId) } -func InsertWithCtxParallel[T AllowedIds, I Insertable[T]](g *errgroup.Group, ctx context.Context, tx *sqlx.Tx, data I, relId InsertId[T]) { +func InsertWithCtxParallel[T AllowedIds, I Insertable[T]](g *errgroup.Group, tx *sqlx.Tx, data I, relId InsertId[T]) { g.Go(func() error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -237,3 +238,11 @@ func IsErrUniqueConstraint(err error) { } } } + +var modelInsertLock sync.Map + +func GetMutex(id string) *sync.Mutex { + val, _ := modelInsertLock.LoadOrStore(id, &sync.Mutex{}) + + return val.(*sync.Mutex) +} diff --git a/internal/foundry/models/db/world.go b/internal/foundry/models/db/world.go index 16558be..300eb50 100644 --- a/internal/foundry/models/db/world.go +++ b/internal/foundry/models/db/world.go @@ -61,10 +61,10 @@ func (w *World) Query(data *InsertId[string]) { func (w *World) InsertObjects(tx *sqlx.Tx) error { relId := InsertId[string]{id: w.ID, fieldName: "world_id"} - group, ctx := errgroup.WithContext(context.Background()) + group, _ := errgroup.WithContext(context.Background()) - InsertWithCtxParallel(group, ctx, tx, w.Compatibility, relId) - InsertWithCtxParallel(group, ctx, tx, w.Relationships, relId) + InsertWithCtxParallel(group, tx, w.Compatibility, relId) + InsertWithCtxParallel(group, tx, w.Relationships, relId) esModulesRelId := &InsertId[string]{id: w.ID, fieldName: "world_id", tableName: "es_modules"} InsertSimpleSliceParallel(group, tx, w.Esmodules, esModulesRelId) diff --git a/internal/foundry/transport/transport.go b/internal/foundry/transport/transport.go index f63bf41..2ce9d1b 100644 --- a/internal/foundry/transport/transport.go +++ b/internal/foundry/transport/transport.go @@ -64,25 +64,26 @@ func NewFoundryTransport(data *FoundryTransportData) *FoundryTransport { } } -func (t *FoundryTransport) CloseMsgChannel(id int, timeout time.Duration) bool { +func (t *FoundryTransport) CloseMsgChannel(msgChan chan []byte, id int, timeout time.Duration) bool { time.Sleep(timeout) t.ChanMutex.Lock() defer t.ChanMutex.Unlock() ok := true - if _, ok = t.ExchangeChan.Msgs[id]; !ok { - return false - } select { - case _, ok = <-t.ExchangeChan.Msgs[id]: + case _, ok = <-msgChan: if ok { - close(t.ExchangeChan.Msgs[id]) - delete(t.ExchangeChan.Msgs, id) + if msgChan == t.ExchangeChan.Msgs[id] { + delete(t.ExchangeChan.Msgs, id) + } + close(msgChan) } default: - close(t.ExchangeChan.Msgs[id]) - delete(t.ExchangeChan.Msgs, id) + if msgChan == t.ExchangeChan.Msgs[id] { + delete(t.ExchangeChan.Msgs, id) + } + close(msgChan) } return ok diff --git a/internal/foundry/transport/websocket.go b/internal/foundry/transport/websocket.go index 31100b6..b338796 100644 --- a/internal/foundry/transport/websocket.go +++ b/internal/foundry/transport/websocket.go @@ -1,6 +1,7 @@ package transport import ( + "context" "encoding/json" "fmt" "net/http" @@ -61,8 +62,8 @@ func (tr *FoundryTransport) HandleWebsocketRequest(msg *types.WsMessage) ([]byte } func (tr *FoundryTransport) ReceiveMessage(id int) ([]byte, error) { - // ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - // defer cancel() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() for { select { @@ -71,14 +72,14 @@ func (tr *FoundryTransport) ReceiveMessage(id int) ([]byte, error) { time.Sleep(5 * time.Microsecond) continue } - tr.CloseMsgChannel(id, 0) + tr.CloseMsgChannel(tr.ExchangeChan.Msgs[id], id, 0) return msg, nil - // case <-ctx.Done(): - // err := ctx.Err() - // if err != nil { - // return nil, err - // } - // return nil, ErrorTimeout + case <-ctx.Done(): + err := ctx.Err() + if err != nil { + return nil, err + } + return nil, ErrorTimeout default: time.Sleep(5 * time.Microsecond) }