diff --git a/internal/store/postgres/kyc_repository.go b/internal/store/postgres/kyc_repository.go index 55b40c8e1..2b4077dca 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() @@ -76,70 +76,58 @@ 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, - }, - goqu.Ex{ - TABLE_ORGANIZATIONS + ".id": input.OrgID, - }, - ). - 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 query with subquery for org name - orgNameSubquery := dialect.From(TABLE_ORGANIZATIONS). - Select("title"). - 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() - 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) } 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 } @@ -177,8 +165,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 } @@ -187,10 +177,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 { @@ -217,7 +207,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 f098295ae..723d5f37c 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" @@ -300,6 +301,208 @@ 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().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().ErrorContains(s.insertKycRow(orgID), "uq_organizations_kyc_org_id_live") + }) +} + +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 (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 (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)) } + +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) + }) +} + +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) + }) +} 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;