From 552ea1b86ea0f1ee53f3403d761ad512b53a1c47 Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 17:19:15 +0530 Subject: [PATCH 1/8] feat(store): kyc org id is unique over live rows only --- .../store/postgres/kyc_repository_test.go | 24 +++++++++++++++++++ ...1012100000_kyc_org_id_live_unique.down.sql | 2 ++ ...261012100000_kyc_org_id_live_unique.up.sql | 2 ++ 3 files changed, 28 insertions(+) create mode 100644 internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.down.sql create mode 100644 internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.up.sql diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index f098295ae..95715d732 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -300,6 +300,30 @@ func (s *OrgKycRepositoryTestSuite) TestList() { } } +func (s *OrgKycRepositoryTestSuite) softDeleteKycRow(orgID string) { + _, err := s.client.ExecContext(s.ctx, "UPDATE organizations_kyc SET deleted_at = now() WHERE org_id = $1 AND deleted_at IS NULL", orgID) + s.Require().NoError(err) +} + +func (s *OrgKycRepositoryTestSuite) insertKycRow(orgID string) error { + _, err := s.client.ExecContext(s.ctx, "INSERT INTO organizations_kyc (org_id, status, link) VALUES ($1, false, '')", orgID) + return err +} + +func (s *OrgKycRepositoryTestSuite) TestOrgIDIsUniqueOverLiveRowsOnly() { + orgID := s.kycs[0].OrgID + + s.Run("a live row cannot be duplicated", func() { + s.Assert().Error(s.insertKycRow(orgID)) + }) + + s.Run("a soft deleted row and a live row coexist", func() { + s.softDeleteKycRow(orgID) + s.Require().NoError(s.insertKycRow(orgID)) + s.Assert().Error(s.insertKycRow(orgID)) + }) +} + func TestOrganizationKYCRepository(t *testing.T) { suite.Run(t, new(OrgKycRepositoryTestSuite)) } diff --git a/internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.down.sql b/internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.down.sql new file mode 100644 index 000000000..ed94d95a8 --- /dev/null +++ b/internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE organizations_kyc ADD CONSTRAINT organizations_kyc_org_id_key UNIQUE (org_id); +DROP INDEX IF EXISTS uq_organizations_kyc_org_id_live; diff --git a/internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.up.sql b/internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.up.sql new file mode 100644 index 000000000..834c4cf7a --- /dev/null +++ b/internal/store/postgres/migrations/20261012100000_kyc_org_id_live_unique.up.sql @@ -0,0 +1,2 @@ +CREATE UNIQUE INDEX IF NOT EXISTS uq_organizations_kyc_org_id_live ON organizations_kyc (org_id) WHERE deleted_at IS NULL; +ALTER TABLE organizations_kyc DROP CONSTRAINT IF EXISTS organizations_kyc_org_id_key; From fb23cef66284cb6ac969299820054ca353743c77 Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 17:42:07 +0530 Subject: [PATCH 2/8] feat(store): kyc reads skip soft-deleted rows --- internal/store/postgres/kyc_repository.go | 2 +- internal/store/postgres/kyc_repository_test.go | 8 ++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 55b40c8e1..4db71fabe 100644 --- a/internal/store/postgres/kyc_repository.go +++ b/internal/store/postgres/kyc_repository.go @@ -48,7 +48,7 @@ func NewOrgKycRepository(dbc *db.Client) *OrgKycRepository { } func (r OrgKycRepository) GetByOrgID(ctx context.Context, orgID string) (kyc.KYC, error) { - query, params, err := dialect.From(TABLE_ORGANIZATIONS_KYC).Where(goqu.Ex{ + query, params, err := fromLive(TABLE_ORGANIZATIONS_KYC).Where(goqu.Ex{ "org_id": orgID, }).ToSQL() diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index 95715d732..b6a63b230 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -324,6 +324,14 @@ func (s *OrgKycRepositoryTestSuite) TestOrgIDIsUniqueOverLiveRowsOnly() { }) } +func (s *OrgKycRepositoryTestSuite) TestGetByOrgIDSkipsSoftDeletedRow() { + orgID := s.kycs[0].OrgID + s.softDeleteKycRow(orgID) + + _, err := s.repository.GetByOrgID(s.ctx, orgID) + s.Assert().ErrorIs(err, kyc.ErrNotExist) +} + func TestOrganizationKYCRepository(t *testing.T) { suite.Run(t, new(OrgKycRepositoryTestSuite)) } From 408c3a13f0ca0b9a9745df4ba99384fe0e7e5131 Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 17:43:25 +0530 Subject: [PATCH 3/8] feat(store): kyc delete keeps the row --- internal/store/postgres/kyc_repository.go | 7 +- .../store/postgres/kyc_repository_test.go | 75 +++++++++++++++++++ 2 files changed, 79 insertions(+), 3 deletions(-) diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 4db71fabe..033a08e78 100644 --- a/internal/store/postgres/kyc_repository.go +++ b/internal/store/postgres/kyc_repository.go @@ -99,6 +99,7 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e goqu.Ex{ TABLE_ORGANIZATIONS_KYC + ".org_id": input.OrgID, }, + live(TABLE_ORGANIZATIONS_KYC), goqu.Ex{ TABLE_ORGANIZATIONS + ".id": input.OrgID, }, @@ -187,10 +188,10 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e return result.KYC.transformToKyc() } -// Delete removes the kyc record of an org. Deleting is idempotent: an org -// without a kyc record returns success. +// Delete marks the kyc record of an org as deleted and keeps the row. Deleting +// is idempotent: an org without a live kyc record returns success. func (r OrgKycRepository) Delete(ctx context.Context, orgID string) error { - query, params, err := dialect.Delete(TABLE_ORGANIZATIONS_KYC).Where(goqu.Ex{ + query, params, err := softDelete(TABLE_ORGANIZATIONS_KYC).Where(goqu.Ex{ "org_id": orgID, }).ToSQL() if err != nil { diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index b6a63b230..f26059ad9 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -2,6 +2,7 @@ package postgres_test import ( "context" + "database/sql" "fmt" "sort" "testing" @@ -332,6 +333,80 @@ func (s *OrgKycRepositoryTestSuite) TestGetByOrgIDSkipsSoftDeletedRow() { s.Assert().ErrorIs(err, kyc.ErrNotExist) } +func (s *OrgKycRepositoryTestSuite) deletedAtOf(orgID string) []sql.NullTime { + var deletedAts []sql.NullTime + err := s.client.SelectContext(s.ctx, &deletedAts, "SELECT deleted_at FROM organizations_kyc WHERE org_id = $1 ORDER BY created_at", orgID) + s.Require().NoError(err) + return deletedAts +} + +func (s *OrgKycRepositoryTestSuite) orgWithKyc(name string) string { + org, err := s.orgRepository.Create(s.ctx, organization.Organization{Name: name}) + s.Require().NoError(err) + _, err = s.repository.Upsert(s.ctx, kyc.KYC{OrgID: org.ID, Status: true, Link: "abcd"}) + s.Require().NoError(err) + return org.ID +} + +func (s *OrgKycRepositoryTestSuite) TestDelete() { + s.Run("keeps the row with deleted_at set", func() { + orgID := s.orgWithKyc("org-delete-keeps-row") + + s.Require().NoError(s.repository.Delete(s.ctx, orgID)) + + deletedAts := s.deletedAtOf(orgID) + s.Require().Len(deletedAts, 1) + s.Assert().True(deletedAts[0].Valid) + }) + + s.Run("a second delete keeps the first deleted_at and returns nil", func() { + orgID := s.orgWithKyc("org-second-delete") + s.Require().NoError(s.repository.Delete(s.ctx, orgID)) + first := s.deletedAtOf(orgID) + + s.Require().NoError(s.repository.Delete(s.ctx, orgID)) + + second := s.deletedAtOf(orgID) + s.Require().Len(second, 1) + s.Assert().True(first[0].Time.Equal(second[0].Time)) + }) + + s.Run("an org without kyc returns nil", func() { + org, err := s.orgRepository.Create(s.ctx, organization.Organization{Name: "org-without-kyc"}) + s.Require().NoError(err) + + s.Assert().NoError(s.repository.Delete(s.ctx, org.ID)) + }) + + s.Run("upsert after delete inserts a new live row", func() { + orgID := s.orgWithKyc("org-upsert-after-delete") + s.Require().NoError(s.repository.Delete(s.ctx, orgID)) + + got, err := s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: true, Link: "fresh-link"}) + + s.Require().NoError(err) + s.Assert().Equal("fresh-link", got.Link) + deletedAts := s.deletedAtOf(orgID) + s.Require().Len(deletedAts, 2) + s.Assert().True(deletedAts[0].Valid) + s.Assert().False(deletedAts[1].Valid) + }) + + s.Run("updating a live row leaves the deleted row untouched", func() { + orgID := s.orgWithKyc("org-update-after-delete") + s.Require().NoError(s.repository.Delete(s.ctx, orgID)) + _, err := s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: true, Link: "fresh-link"}) + s.Require().NoError(err) + + _, err = s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: true, Link: "newer-link"}) + s.Require().NoError(err) + + var links []string + s.Require().NoError(s.client.SelectContext(s.ctx, &links, "SELECT link FROM organizations_kyc WHERE org_id = $1 AND deleted_at IS NOT NULL", orgID)) + s.Assert().Equal([]string{"abcd"}, links) + }) +} + func TestOrganizationKYCRepository(t *testing.T) { suite.Run(t, new(OrgKycRepositoryTestSuite)) } From 319de71896f3c6cd53e9036187301b330517ec65 Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 17:43:52 +0530 Subject: [PATCH 4/8] feat(store): kyc list ignores a soft-deleted kyc row --- internal/store/postgres/kyc_repository.go | 2 +- .../store/postgres/kyc_repository_test.go | 20 +++++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 033a08e78..075c5e479 100644 --- a/internal/store/postgres/kyc_repository.go +++ b/internal/store/postgres/kyc_repository.go @@ -218,7 +218,7 @@ func (r OrgKycRepository) List(ctx context.Context) ([]kyc.KYC, error) { // Build query with join condition and COALESCE expressions query, params, err := dialect.From(orgs). - LeftJoin(orgKycs, goqu.On(orgs.Col(COLUMN_ID).Eq(orgKycs.Col(COLUMN_ORG_ID)))). + LeftJoin(orgKycs, goqu.On(orgs.Col(COLUMN_ID).Eq(orgKycs.Col(COLUMN_ORG_ID)), live(TABLE_ORGANIZATIONS_KYC))). Select( orgs.Col(COLUMN_ID), goqu.COALESCE(orgKycs.Col(COLUMN_STATUS), false).As(COLUMN_STATUS), diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index f26059ad9..f193729fd 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -407,6 +407,26 @@ func (s *OrgKycRepositoryTestSuite) TestDelete() { }) } +func (s *OrgKycRepositoryTestSuite) TestListIgnoresSoftDeletedKyc() { + s.Require().NoError(s.cleanup()) + deletedOrgID := s.orgWithKyc("org-with-deleted-kyc") + liveOrgID := s.orgWithKyc("org-with-live-kyc") + s.Require().NoError(s.repository.Delete(s.ctx, deletedOrgID)) + + got, err := s.repository.List(s.ctx) + + s.Require().NoError(err) + byOrg := map[string]kyc.KYC{} + for _, k := range got { + byOrg[k.OrgID] = k + } + s.Require().Len(got, 2) + s.Assert().False(byOrg[deletedOrgID].Status) + s.Assert().Equal("", byOrg[deletedOrgID].Link) + s.Assert().True(byOrg[liveOrgID].Status) + s.Assert().Equal("abcd", byOrg[liveOrgID].Link) +} + func TestOrganizationKYCRepository(t *testing.T) { suite.Run(t, new(OrgKycRepositoryTestSuite)) } From 5a09640b172ba093083d2a343d91cbf79d2a7a90 Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 18:11:19 +0530 Subject: [PATCH 5/8] fix(store): kyc upsert refuses an org that is soft-deleted --- internal/store/postgres/kyc_repository.go | 33 +++++++++++- .../store/postgres/kyc_repository_test.go | 51 +++++++++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 075c5e479..4f7664b36 100644 --- a/internal/store/postgres/kyc_repository.go +++ b/internal/store/postgres/kyc_repository.go @@ -86,6 +86,10 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e } var result kycWithOrgName + if err := r.ensureLiveOrg(ctx, input.OrgID); err != nil { + return kyc.KYC{}, err + } + _, err := r.GetByOrgID(ctx, input.OrgID) if err == nil { // kyc for org exists, prepare UPDATE query with JOIN @@ -103,6 +107,7 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e goqu.Ex{ TABLE_ORGANIZATIONS + ".id": input.OrgID, }, + live(TABLE_ORGANIZATIONS), ). Returning( goqu.I(TABLE_ORGANIZATIONS_KYC+".*"), @@ -113,7 +118,7 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e } } else if err.Error() == kyc.ErrNotExist.Error() { // kyc for org doesn't exist, prepare INSERT query with subquery for org name - orgNameSubquery := dialect.From(TABLE_ORGANIZATIONS). + orgNameSubquery := fromLive(TABLE_ORGANIZATIONS). Select("title"). Where(goqu.Ex{"id": input.OrgID}) @@ -188,6 +193,32 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e return result.KYC.transformToKyc() } +func (r OrgKycRepository) ensureLiveOrg(ctx context.Context, orgID string) error { + query, params, err := fromLive(TABLE_ORGANIZATIONS). + Select(goqu.L("1")). + Where(goqu.Ex{"id": orgID}). + ToSQL() + if err != nil { + return fmt.Errorf("%w: %w", errQuery, err) + } + + var found int + err = r.dbc.WithTimeout(ctx, TABLE_ORGANIZATIONS, "EnsureLiveOrg", func(ctx context.Context) error { + return r.dbc.GetContext(ctx, &found, query, params...) + }) + err = checkPostgresError(err) + switch { + case err == nil: + return nil + case errors.Is(err, sql.ErrNoRows): + return kyc.ErrOrgDoesntExist + case errors.Is(err, ErrInvalidTextRepresentation): + return kyc.ErrInvalidUUID + default: + return err + } +} + // Delete marks the kyc record of an org as deleted and keeps the row. Deleting // is idempotent: an org without a live kyc record returns success. func (r OrgKycRepository) Delete(ctx context.Context, orgID string) error { diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index f193729fd..fddaa3d06 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -430,3 +430,54 @@ func (s *OrgKycRepositoryTestSuite) TestListIgnoresSoftDeletedKyc() { func TestOrganizationKYCRepository(t *testing.T) { suite.Run(t, new(OrgKycRepositoryTestSuite)) } + +func (s *OrgKycRepositoryTestSuite) softDeleteOrg(orgID string) { + _, err := s.client.ExecContext(s.ctx, "UPDATE organizations SET deleted_at = now() WHERE id = $1", orgID) + s.Require().NoError(err) +} + +func (s *OrgKycRepositoryTestSuite) kycAuditCount(orgID string) int { + var n int + err := s.client.QueryRowxContext(s.ctx, fmt.Sprintf( + "SELECT count(*) FROM %s WHERE target_id = $1 AND event IN ('kyc.verified', 'kyc.unverified')", postgres.TABLE_AUDITRECORDS), + orgID).Scan(&n) + s.Require().NoError(err) + return n +} + +func (s *OrgKycRepositoryTestSuite) TestUpsertRefusesSoftDeletedOrg() { + s.Run("first time insert writes nothing", func() { + org, err := s.orgRepository.Create(s.ctx, organization.Organization{Name: "deleted-org-no-kyc"}) + s.Require().NoError(err) + s.softDeleteOrg(org.ID) + + _, err = s.repository.Upsert(s.ctx, kyc.KYC{OrgID: org.ID, Status: true, Link: "abcd"}) + + s.Assert().ErrorIs(err, kyc.ErrOrgDoesntExist) + s.Assert().Empty(s.deletedAtOf(org.ID)) + s.Assert().Equal(0, s.kycAuditCount(org.ID)) + }) + + s.Run("existing kyc row stays unchanged", func() { + orgID := s.orgWithKyc("deleted-org-with-kyc") + auditsBefore := s.kycAuditCount(orgID) + s.softDeleteOrg(orgID) + + _, err := s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: false, Link: "changed"}) + + s.Assert().ErrorIs(err, kyc.ErrOrgDoesntExist) + var status bool + var link string + s.Require().NoError(s.client.QueryRowxContext(s.ctx, "SELECT status, link FROM organizations_kyc WHERE org_id = $1", orgID).Scan(&status, &link)) + s.Assert().True(status) + s.Assert().Equal("abcd", link) + s.Assert().Equal(auditsBefore, s.kycAuditCount(orgID)) + }) + + s.Run("live org still works", func() { + orgID := s.orgWithKyc("live-org") + got, err := s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: true, Link: "new-link"}) + s.Require().NoError(err) + s.Assert().Equal("new-link", got.Link) + }) +} From b7ee7979eb89720f166f710f1175c42a1d8cb15f Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 18:40:52 +0530 Subject: [PATCH 6/8] test(store): the kyc unique index test pins the index name --- internal/store/postgres/kyc_repository_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index fddaa3d06..72dd0fee1 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -315,13 +315,13 @@ func (s *OrgKycRepositoryTestSuite) TestOrgIDIsUniqueOverLiveRowsOnly() { orgID := s.kycs[0].OrgID s.Run("a live row cannot be duplicated", func() { - s.Assert().Error(s.insertKycRow(orgID)) + s.Assert().ErrorContains(s.insertKycRow(orgID), "uq_organizations_kyc_org_id_live") }) s.Run("a soft deleted row and a live row coexist", func() { s.softDeleteKycRow(orgID) s.Require().NoError(s.insertKycRow(orgID)) - s.Assert().Error(s.insertKycRow(orgID)) + s.Assert().ErrorContains(s.insertKycRow(orgID), "uq_organizations_kyc_org_id_live") }) } From 6119be532a9cbc5cd4e97a7bc77aad6af2fb7276 Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Mon, 12 Oct 2026 18:42:44 +0530 Subject: [PATCH 7/8] fix(store): kyc upsert decides the org is live in the same statement it writes --- internal/store/postgres/kyc_repository.go | 64 +++++++---------------- 1 file changed, 19 insertions(+), 45 deletions(-) diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 4f7664b36..3a725d4a3 100644 --- a/internal/store/postgres/kyc_repository.go +++ b/internal/store/postgres/kyc_repository.go @@ -86,10 +86,6 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e } var result kycWithOrgName - if err := r.ensureLiveOrg(ctx, input.OrgID); err != nil { - return kyc.KYC{}, err - } - _, err := r.GetByOrgID(ctx, input.OrgID) if err == nil { // kyc for org exists, prepare UPDATE query with JOIN @@ -117,21 +113,20 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e return kyc.KYC{}, fmt.Errorf("%w: %w", errQuery, err) } } else if err.Error() == kyc.ErrNotExist.Error() { - // kyc for org doesn't exist, prepare INSERT query with subquery for org name - orgNameSubquery := fromLive(TABLE_ORGANIZATIONS). - Select("title"). + // kyc for org doesn't exist, prepare INSERT from the live org row + liveOrg := fromLive(TABLE_ORGANIZATIONS). + Select(goqu.I("id"), goqu.V(input.Status), goqu.V(input.Link)). Where(goqu.Ex{"id": input.OrgID}) - - query, params, err = dialect.Insert(TABLE_ORGANIZATIONS_KYC). - Rows(goqu.Record{ - "org_id": input.OrgID, - "status": input.Status, - "link": input.Link, - }). - Returning( - goqu.I(TABLE_ORGANIZATIONS_KYC+".*"), - orgNameSubquery.As("org_name"), - ).ToSQL() + inserted := dialect.Insert(TABLE_ORGANIZATIONS_KYC). + Cols("org_id", "status", "link"). + FromQuery(liveOrg). + Returning(goqu.I(TABLE_ORGANIZATIONS_KYC + ".*")) + + query, params, err = dialect.From("inserted"). + With("inserted", inserted). + Join(goqu.T(TABLE_ORGANIZATIONS), goqu.On(goqu.I(TABLE_ORGANIZATIONS+".id").Eq(goqu.I("inserted.org_id")))). + Select(goqu.I("inserted.*"), goqu.I(TABLE_ORGANIZATIONS+".title").As("org_name")). + ToSQL() if err != nil { return kyc.KYC{}, fmt.Errorf("%w: %w", errQuery, err) } @@ -146,6 +141,9 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e if err = r.dbc.WithTxn(ctx, sql.TxOptions{}, func(tx *sqlx.Tx) error { return r.dbc.WithTimeout(ctx, TABLE_ORGANIZATIONS_KYC, "Upsert", func(ctx context.Context) error { if err := tx.QueryRowxContext(ctx, query, params...).StructScan(&result); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return kyc.ErrOrgDoesntExist + } return err } @@ -183,8 +181,10 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e switch { case errors.Is(err, ErrCheckViolation): return kyc.KYC{}, kyc.ErrKycLinkNotSet - case errors.Is(err, ErrForeignKeyViolation): + case errors.Is(err, ErrForeignKeyViolation), errors.Is(err, kyc.ErrOrgDoesntExist): return kyc.KYC{}, kyc.ErrOrgDoesntExist + case errors.Is(err, ErrInvalidTextRepresentation): + return kyc.KYC{}, kyc.ErrInvalidUUID default: return kyc.KYC{}, err } @@ -193,32 +193,6 @@ func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, e return result.KYC.transformToKyc() } -func (r OrgKycRepository) ensureLiveOrg(ctx context.Context, orgID string) error { - query, params, err := fromLive(TABLE_ORGANIZATIONS). - Select(goqu.L("1")). - Where(goqu.Ex{"id": orgID}). - ToSQL() - if err != nil { - return fmt.Errorf("%w: %w", errQuery, err) - } - - var found int - err = r.dbc.WithTimeout(ctx, TABLE_ORGANIZATIONS, "EnsureLiveOrg", func(ctx context.Context) error { - return r.dbc.GetContext(ctx, &found, query, params...) - }) - err = checkPostgresError(err) - switch { - case err == nil: - return nil - case errors.Is(err, sql.ErrNoRows): - return kyc.ErrOrgDoesntExist - case errors.Is(err, ErrInvalidTextRepresentation): - return kyc.ErrInvalidUUID - default: - return err - } -} - // Delete marks the kyc record of an org as deleted and keeps the row. Deleting // is idempotent: an org without a live kyc record returns success. func (r OrgKycRepository) Delete(ctx context.Context, orgID string) error { From a5f19be67bb55a0d4cc537722a28f6504819b7fe Mon Sep 17 00:00:00 2001 From: Rohil Surana Date: Tue, 13 Oct 2026 11:38:46 +0530 Subject: [PATCH 8/8] fix(store): kyc upsert picks update or insert inside one statement --- internal/store/postgres/kyc_repository.go | 90 ++++++++----------- .../store/postgres/kyc_repository_test.go | 25 ++++++ 2 files changed, 62 insertions(+), 53 deletions(-) diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 3a725d4a3..2b4077dca 100644 --- a/internal/store/postgres/kyc_repository.go +++ b/internal/store/postgres/kyc_repository.go @@ -76,65 +76,49 @@ func (r OrgKycRepository) GetByOrgID(ctx context.Context, orgID string) (kyc.KYC } func (r OrgKycRepository) Upsert(ctx context.Context, input kyc.KYC) (kyc.KYC, error) { - var query string - var params []any - - // Struct to hold KYC data + org name type kycWithOrgName struct { KYC OrgName string `db:"org_name"` } var result kycWithOrgName - _, err := r.GetByOrgID(ctx, input.OrgID) - if err == nil { - // kyc for org exists, prepare UPDATE query with JOIN - query, params, err = dialect.Update(TABLE_ORGANIZATIONS_KYC). - Set(goqu.Record{ - "status": input.Status, - "link": input.Link, - }). - From(TABLE_ORGANIZATIONS). - Where( - goqu.Ex{ - TABLE_ORGANIZATIONS_KYC + ".org_id": input.OrgID, - }, - live(TABLE_ORGANIZATIONS_KYC), - goqu.Ex{ - TABLE_ORGANIZATIONS + ".id": input.OrgID, - }, - live(TABLE_ORGANIZATIONS), - ). - Returning( - goqu.I(TABLE_ORGANIZATIONS_KYC+".*"), - goqu.I(TABLE_ORGANIZATIONS+".title").As("org_name"), - ).ToSQL() - if err != nil { - return kyc.KYC{}, fmt.Errorf("%w: %w", errQuery, err) - } - } else if err.Error() == kyc.ErrNotExist.Error() { - // kyc for org doesn't exist, prepare INSERT from the live org row - liveOrg := fromLive(TABLE_ORGANIZATIONS). - Select(goqu.I("id"), goqu.V(input.Status), goqu.V(input.Link)). - Where(goqu.Ex{"id": input.OrgID}) - inserted := dialect.Insert(TABLE_ORGANIZATIONS_KYC). - Cols("org_id", "status", "link"). - FromQuery(liveOrg). - Returning(goqu.I(TABLE_ORGANIZATIONS_KYC + ".*")) - - query, params, err = dialect.From("inserted"). - With("inserted", inserted). - Join(goqu.T(TABLE_ORGANIZATIONS), goqu.On(goqu.I(TABLE_ORGANIZATIONS+".id").Eq(goqu.I("inserted.org_id")))). - Select(goqu.I("inserted.*"), goqu.I(TABLE_ORGANIZATIONS+".title").As("org_name")). - ToSQL() - if err != nil { - return kyc.KYC{}, fmt.Errorf("%w: %w", errQuery, err) - } - } else if errors.Is(err, kyc.ErrInvalidUUID) { - // invalid UUID provided - return kyc.KYC{}, kyc.ErrInvalidUUID - } else { - // unexpected error happened in getting org kyc + updated := dialect.Update(TABLE_ORGANIZATIONS_KYC). + Set(goqu.Record{ + "status": input.Status, + "link": input.Link, + }). + From(TABLE_ORGANIZATIONS). + Where( + goqu.Ex{ + TABLE_ORGANIZATIONS_KYC + ".org_id": input.OrgID, + TABLE_ORGANIZATIONS + ".id": input.OrgID, + }, + live(TABLE_ORGANIZATIONS_KYC), + live(TABLE_ORGANIZATIONS), + ). + Returning(goqu.I(TABLE_ORGANIZATIONS_KYC + ".*")) + + liveOrgWithoutUpdate := fromLive(TABLE_ORGANIZATIONS). + Select(goqu.I("id"), goqu.V(input.Status), goqu.V(input.Link)). + Where( + goqu.Ex{"id": input.OrgID}, + goqu.L("NOT EXISTS (SELECT 1 FROM updated)"), + ) + inserted := dialect.Insert(TABLE_ORGANIZATIONS_KYC). + Cols("org_id", "status", "link"). + FromQuery(liveOrgWithoutUpdate). + Returning(goqu.I(TABLE_ORGANIZATIONS_KYC + ".*")) + + written := dialect.From("updated").Select(goqu.Star()). + UnionAll(dialect.From("inserted").Select(goqu.Star())) + + query, params, err := dialect.From(written.As("written")). + With("updated", updated). + With("inserted", inserted). + Join(goqu.T(TABLE_ORGANIZATIONS), goqu.On(goqu.I(TABLE_ORGANIZATIONS+".id").Eq(goqu.I("written.org_id")))). + Select(goqu.I("written.*"), goqu.I(TABLE_ORGANIZATIONS+".title").As("org_name")). + ToSQL() + if err != nil { return kyc.KYC{}, fmt.Errorf("%w: %w", errQuery, err) } diff --git a/internal/store/postgres/kyc_repository_test.go b/internal/store/postgres/kyc_repository_test.go index 72dd0fee1..723d5f37c 100644 --- a/internal/store/postgres/kyc_repository_test.go +++ b/internal/store/postgres/kyc_repository_test.go @@ -481,3 +481,28 @@ func (s *OrgKycRepositoryTestSuite) TestUpsertRefusesSoftDeletedOrg() { s.Assert().Equal("new-link", got.Link) }) } + +func (s *OrgKycRepositoryTestSuite) TestUpsertWritesExactlyOneLiveRow() { + s.Run("updating a live row adds no row", func() { + orgID := s.orgWithKyc("upsert-update-org") + + _, err := s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: false, Link: "changed"}) + + s.Require().NoError(err) + s.Assert().Len(s.deletedAtOf(orgID), 1) + }) + + s.Run("a deleted row is kept and a new live row is added", func() { + orgID := s.orgWithKyc("upsert-after-delete-org") + s.softDeleteKycRow(orgID) + + got, err := s.repository.Upsert(s.ctx, kyc.KYC{OrgID: orgID, Status: true, Link: "again"}) + + s.Require().NoError(err) + s.Assert().Equal("again", got.Link) + deletedAts := s.deletedAtOf(orgID) + s.Require().Len(deletedAts, 2) + s.Assert().True(deletedAts[0].Valid) + s.Assert().False(deletedAts[1].Valid) + }) +}