From a95e44cf42df7906e4920931807ed2af84ef455a Mon Sep 17 00:00:00 2001 From: Eli Ribble Date: Mon, 4 May 2026 20:57:50 +0000 Subject: [PATCH] Use transactions to set the communication status changes Not doing it yet, but soon we'll do log entries for them. --- db/connection.go | 32 ++++++++++++++++++++++++++++-- db/query/arcgis/account.go | 2 +- db/query/arcgis/service_feature.go | 2 +- db/query/arcgis/service_map.go | 2 +- db/query/arcgis/user.go | 2 +- db/query/arcgis/user_privileges.go | 4 ++-- db/query/public/communication.go | 16 +++++++-------- platform/communication.go | 27 +++++++++++++++++++++---- 8 files changed, 67 insertions(+), 20 deletions(-) diff --git a/db/connection.go b/db/connection.go index 8f207e37..dd35cc53 100644 --- a/db/connection.go +++ b/db/connection.go @@ -34,13 +34,24 @@ var ( pgOnce sync.Once ) +type Tx = pgx.Tx + +func BeginTxn(ctx context.Context) (pgx.Tx, error) { + return PGInstance.PGXPool.BeginTx(ctx, pgx.TxOptions{}) +} func ExecuteNone(ctx context.Context, stmt postgres.Statement) error { query, args := stmt.Sql() _, err := PGInstance.PGXPool.Query(ctx, query, args...) return err } -func ExecuteNoneTx(ctx context.Context, txn bob.Tx, stmt postgres.Statement) error { +func ExecuteNoneTx(ctx context.Context, txn Tx, stmt postgres.Statement) error { + query, args := stmt.Sql() + + _, err := txn.Query(ctx, query, args...) + return err +} +func ExecuteNoneTxBob(ctx context.Context, txn bob.Tx, stmt postgres.Statement) error { query, args := stmt.Sql() _, err := txn.QueryContext(ctx, query, args...) @@ -55,7 +66,24 @@ func ExecuteOne[T any](ctx context.Context, stmt postgres.Statement) (*T, error) } return pgx.CollectOneRow(row, pgx.RowToAddrOfStructByPos[T]) } -func ExecuteOneTx[T any](ctx context.Context, txn bob.Tx, stmt postgres.Statement) (*T, error) { +func ExecuteOneTx[T any](ctx context.Context, txn Tx, stmt postgres.Statement) (*T, error) { + query, args := stmt.Sql() + + //result, err := scan.One(ctx, txn, scan.StructMapper[T](), query, args...) + rows, err := txn.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("txn query: %w", err) + } + results, err := pgx.CollectRows(rows, pgx.RowToStructByName[T]) + if err != nil { + return nil, fmt.Errorf("collect rows: %w", err) + } + if len(results) < 1 { + return nil, fmt.Errorf("no results") + } + return &results[0], err +} +func ExecuteOneTxBob[T any](ctx context.Context, txn bob.Tx, stmt postgres.Statement) (*T, error) { query, args := stmt.Sql() result, err := scan.One(ctx, txn, scan.StructMapper[T](), query, args...) diff --git a/db/query/arcgis/account.go b/db/query/arcgis/account.go index c484a978..2a6b5428 100644 --- a/db/query/arcgis/account.go +++ b/db/query/arcgis/account.go @@ -20,5 +20,5 @@ func AccountFromID(ctx context.Context, org_id string) (*model.Account, error) { func AccountInsert(ctx context.Context, txn bob.Tx, m *model.Account) (*model.Account, error) { statement := table.Account.INSERT(table.Account.AllColumns). MODEL(m) - return db.ExecuteOneTx[model.Account](ctx, txn, statement) + return db.ExecuteOneTxBob[model.Account](ctx, txn, statement) } diff --git a/db/query/arcgis/service_feature.go b/db/query/arcgis/service_feature.go index fc9238a2..17359c3d 100644 --- a/db/query/arcgis/service_feature.go +++ b/db/query/arcgis/service_feature.go @@ -27,5 +27,5 @@ func ServiceFeatureFromURL(ctx context.Context, url string) (*model.ServiceFeatu func ServiceFeatureInsert(ctx context.Context, txn bob.Tx, m *model.ServiceFeature) error { statement := table.ServiceMap.INSERT(table.ServiceMap.MutableColumns). MODEL(m) - return db.ExecuteNoneTx(ctx, txn, statement) + return db.ExecuteNoneTxBob(ctx, txn, statement) } diff --git a/db/query/arcgis/service_map.go b/db/query/arcgis/service_map.go index 4a7fe05f..9d25f9c1 100644 --- a/db/query/arcgis/service_map.go +++ b/db/query/arcgis/service_map.go @@ -27,5 +27,5 @@ func ServiceMapsFromAccountID(ctx context.Context, account_id string) ([]*model. func ServiceMapInsert(ctx context.Context, txn bob.Tx, m *model.ServiceMap) error { statement := table.ServiceMap.INSERT(table.ServiceMap.MutableColumns). MODEL(m) - return db.ExecuteNoneTx(ctx, txn, statement) + return db.ExecuteNoneTxBob(ctx, txn, statement) } diff --git a/db/query/arcgis/user.go b/db/query/arcgis/user.go index 33f0f43f..5452285b 100644 --- a/db/query/arcgis/user.go +++ b/db/query/arcgis/user.go @@ -20,5 +20,5 @@ func UserInsert(ctx context.Context, txn bob.Tx, m *model.User) (*model.User, er statement := table.User.INSERT(table.User.MutableColumns). MODEL(m). RETURNING(table.User.AllColumns) - return db.ExecuteOneTx[model.User](ctx, txn, statement) + return db.ExecuteOneTxBob[model.User](ctx, txn, statement) } diff --git a/db/query/arcgis/user_privileges.go b/db/query/arcgis/user_privileges.go index 9947a718..fbac1ece 100644 --- a/db/query/arcgis/user_privileges.go +++ b/db/query/arcgis/user_privileges.go @@ -13,10 +13,10 @@ import ( func UserPrivilegesDeleteByUserID(ctx context.Context, txn bob.Tx, id string) error { statement := table.User.DELETE(). WHERE(table.User.ID.EQ(postgres.String(id))) - return db.ExecuteNoneTx(ctx, txn, statement) + return db.ExecuteNoneTxBob(ctx, txn, statement) } func UserPrivilegeInsert(ctx context.Context, txn bob.Tx, m *model.UserPrivilege) error { statement := table.UserPrivilege.INSERT(table.UserPrivilege.MutableColumns). MODEL(m) - return db.ExecuteNoneTx(ctx, txn, statement) + return db.ExecuteNoneTxBob(ctx, txn, statement) } diff --git a/db/query/public/communication.go b/db/query/public/communication.go index 593d595a..2aa13be3 100644 --- a/db/query/public/communication.go +++ b/db/query/public/communication.go @@ -32,7 +32,7 @@ func CommunicationsFromOrganization(ctx context.Context, org_id int64) ([]*model WHERE(table.Communication.OrganizationID.EQ(postgres.Int(org_id))) return db.ExecuteMany[model.Communication](ctx, statement) } -func CommunicationMarkInvalid(ctx context.Context, org_id int64, user_id int64, comm_id int64) error { +func CommunicationMarkInvalid(ctx context.Context, txn db.Tx, org_id int64, user_id int64, comm_id int64) error { statement := table.Communication.UPDATE(). SET( table.Communication.Invalidated.SET(postgres.TimestampT(time.Now())), @@ -40,9 +40,9 @@ func CommunicationMarkInvalid(ctx context.Context, org_id int64, user_id int64, ). WHERE(table.Communication.OrganizationID.EQ(postgres.Int(org_id)).AND( table.Communication.ID.EQ(postgres.Int(comm_id)))) - return db.ExecuteNone(ctx, statement) + return db.ExecuteNoneTx(ctx, txn, statement) } -func CommunicationMarkPendingResponse(ctx context.Context, org_id int64, user_id int64, comm_id int64) error { +func CommunicationMarkPendingResponse(ctx context.Context, txn db.Tx, org_id int64, user_id int64, comm_id int64) error { statement := table.Communication.UPDATE(). SET( table.Communication.SetPending.SET(postgres.TimestampT(time.Now())), @@ -50,9 +50,9 @@ func CommunicationMarkPendingResponse(ctx context.Context, org_id int64, user_id ). WHERE(table.Communication.OrganizationID.EQ(postgres.Int(org_id)).AND( table.Communication.ID.EQ(postgres.Int(comm_id)))) - return db.ExecuteNone(ctx, statement) + return db.ExecuteNoneTx(ctx, txn, statement) } -func CommunicationMarkPossibleIssue(ctx context.Context, org_id int64, user_id int64, comm_id int64) error { +func CommunicationMarkPossibleIssue(ctx context.Context, txn db.Tx, org_id int64, user_id int64, comm_id int64) error { statement := table.Communication.UPDATE(). SET( table.Communication.SetPossibleIssue.SET(postgres.TimestampT(time.Now())), @@ -60,9 +60,9 @@ func CommunicationMarkPossibleIssue(ctx context.Context, org_id int64, user_id i ). WHERE(table.Communication.OrganizationID.EQ(postgres.Int(org_id)).AND( table.Communication.ID.EQ(postgres.Int(comm_id)))) - return db.ExecuteNone(ctx, statement) + return db.ExecuteNoneTx(ctx, txn, statement) } -func CommunicationMarkPossibleResolved(ctx context.Context, org_id int64, user_id int64, comm_id int64) error { +func CommunicationMarkPossibleResolved(ctx context.Context, txn db.Tx, org_id int64, user_id int64, comm_id int64) error { statement := table.Communication.UPDATE(). SET( table.Communication.SetPossibleResolved.SET(postgres.TimestampT(time.Now())), @@ -70,5 +70,5 @@ func CommunicationMarkPossibleResolved(ctx context.Context, org_id int64, user_i ). WHERE(table.Communication.OrganizationID.EQ(postgres.Int(org_id)).AND( table.Communication.ID.EQ(postgres.Int(comm_id)))) - return db.ExecuteNone(ctx, statement) + return db.ExecuteNoneTx(ctx, txn, statement) } diff --git a/platform/communication.go b/platform/communication.go index ed295598..91d396dc 100644 --- a/platform/communication.go +++ b/platform/communication.go @@ -2,9 +2,12 @@ package platform import ( "context" + "fmt" + "github.com/Gleipnir-Technology/nidus-sync/db" "github.com/Gleipnir-Technology/nidus-sync/db/gen/nidus-sync/public/model" querypublic "github.com/Gleipnir-Technology/nidus-sync/db/query/public" + "github.com/Gleipnir-Technology/nidus-sync/lint" ) func CommunicationsForOrganization(ctx context.Context, org_id int64) ([]*model.Communication, error) { @@ -21,14 +24,30 @@ func CommunicationFromID(ctx context.Context, user User, comm_id int64) (*model. return comm, nil } func CommunicationMarkInvalid(ctx context.Context, user User, comm_id int64) error { - return querypublic.CommunicationMarkInvalid(ctx, int64(user.Organization.ID), int64(user.ID), comm_id) + return communicationMark(ctx, user, comm_id, querypublic.CommunicationMarkInvalid) } func CommunicationMarkPendingResponse(ctx context.Context, user User, comm_id int64) error { - return querypublic.CommunicationMarkPendingResponse(ctx, int64(user.Organization.ID), int64(user.ID), comm_id) + return communicationMark(ctx, user, comm_id, querypublic.CommunicationMarkPendingResponse) } func CommunicationMarkPossibleIssue(ctx context.Context, user User, comm_id int64) error { - return querypublic.CommunicationMarkPossibleIssue(ctx, int64(user.Organization.ID), int64(user.ID), comm_id) + return communicationMark(ctx, user, comm_id, querypublic.CommunicationMarkPossibleIssue) } func CommunicationMarkPossibleResolved(ctx context.Context, user User, comm_id int64) error { - return querypublic.CommunicationMarkPossibleResolved(ctx, int64(user.Organization.ID), int64(user.ID), comm_id) + return communicationMark(ctx, user, comm_id, querypublic.CommunicationMarkPossibleResolved) +} + +type markFunc = func(context.Context, db.Tx, int64, int64, int64) error + +func communicationMark(ctx context.Context, user User, comm_id int64, f markFunc) error { + txn, err := db.BeginTxn(ctx) + if err != nil { + return fmt.Errorf("begin txn: %w", err) + } + defer lint.LogOnErrRollback(txn.Rollback, ctx, "rollback") + err = f(ctx, txn, int64(user.Organization.ID), int64(user.ID), comm_id) + if err != nil { + return fmt.Errorf("mark: %w", err) + } + txn.Commit(ctx) + return nil }