feat: add PENDING_VERIFICATION status to identity domain

api/proto/meridian/identity/v1/identity.proto

@@ -72,6 +72,8 @@ enum IdentityStatus {
   IDENTITY_STATUS_SUSPENDED = 3;
   // IDENTITY_STATUS_LOCKED means the identity has been locked due to excessive failed authentication attempts.
   IDENTITY_STATUS_LOCKED = 4;
+  // IDENTITY_STATUS_PENDING_VERIFICATION means the identity has self-registered but has not yet verified their email address.
+  IDENTITY_STATUS_PENDING_VERIFICATION = 5;
 }
 
 // Role defines the set of predefined roles that can be assigned to an identity.

services/identity/connector/connector.go

@@ -129,6 +129,11 @@ func (c *Connector) Login(ctx context.Context, _ []string, username, password st
 			"tenant_id", tenantID,
 			"identity_id", identity.ID())
 		return Identity{}, false, nil
+	case domain.IdentityStatusPendingVerification:
+		c.logger.InfoContext(ctx, "connector: login rejected — email not yet verified",
+			"tenant_id", tenantID,
+			"identity_id", identity.ID())
+		return Identity{}, false, domain.ErrEmailNotVerified
 	case domain.IdentityStatusActive:
 		// valid — proceed to password verification
 	default:

services/identity/connector/connector_test.go

@@ -520,3 +520,17 @@ func TestLogin_TenantContextPropagation(t *testing.T) {
 		require.Error(t, err)
 	})
 }
+
+func TestLogin_PendingVerificationAccount_ReturnsErrEmailNotVerified(t *testing.T) {
+	id, err := domain.NewSelfRegisteredIdentity(connTestTID, "unverified@example.com", true)
+	require.NoError(t, err)
+
+	repo := &mockRepo{identity: id}
+	c := newConnector(t, repo)
+	ctx := ctxWithTenant(t, "volterra")
+
+	_, valid, err := c.Login(ctx, nil, "unverified@example.com", testPassword)
+
+	assert.False(t, valid)
+	assert.ErrorIs(t, err, domain.ErrEmailNotVerified)
+}

services/identity/domain/errors.go

@@ -20,4 +20,6 @@ var (
 	ErrInsufficientRolePermissions = errors.New("insufficient permissions to grant this role")
 	ErrVersionConflict             = errors.New("version conflict: resource was modified by another transaction")
 	ErrTenantIDRequired            = errors.New("tenant ID is required")
+	ErrNotPendingVerification      = errors.New("identity is not in PENDING_VERIFICATION status")
+	ErrEmailNotVerified            = errors.New("email address has not been verified")
 )

services/identity/domain/identity.go

@@ -13,10 +13,11 @@ type IdentityStatus string
 
 // Identity status constants
 const (
-	IdentityStatusPendingInvite IdentityStatus = "PENDING_INVITE"
-	IdentityStatusActive        IdentityStatus = "ACTIVE"
-	IdentityStatusSuspended     IdentityStatus = "SUSPENDED"
-	IdentityStatusLocked        IdentityStatus = "LOCKED"
+	IdentityStatusPendingInvite       IdentityStatus = "PENDING_INVITE"
+	IdentityStatusPendingVerification IdentityStatus = "PENDING_VERIFICATION"
+	IdentityStatusActive              IdentityStatus = "ACTIVE"
+	IdentityStatusSuspended           IdentityStatus = "SUSPENDED"
+	IdentityStatusLocked              IdentityStatus = "LOCKED"
 )
 
 // maxFailedAttempts is the number of consecutive failed login attempts before lockout.
@@ -68,6 +69,34 @@ func NewIdentity(tenantID tenant.TenantID, email string) (*Identity, error) {
 	}, nil
 }
 
+// NewSelfRegisteredIdentity creates a new identity for a self-registration flow.
+// When verificationRequired is true the identity starts in PENDING_VERIFICATION status;
+// otherwise it starts in ACTIVE status (e.g. when email is already trusted).
+func NewSelfRegisteredIdentity(tenantID tenant.TenantID, email string, verificationRequired bool) (*Identity, error) {
+	if tenantID.IsEmpty() {
+		return nil, ErrTenantIDRequired
+	}
+	if !emailRegex.MatchString(email) {
+		return nil, ErrInvalidEmail
+	}
+
+	status := IdentityStatusActive
+	if verificationRequired {
+		status = IdentityStatusPendingVerification
+	}
+
+	now := time.Now()
+	return &Identity{
+		id:        uuid.New(),
+		tenantID:  tenantID,
+		email:     email,
+		status:    status,
+		createdAt: now,
+		updatedAt: now,
+		version:   1,
+	}, nil
+}
+
 // ReconstructIdentity recreates an Identity from persistence layer data.
 // This should only be used by repositories when loading from the database.
 // baseVersion is set to version so the repository can detect concurrent modifications.
@@ -186,11 +215,10 @@ func (i *Identity) Activate() error {
 		i.updatedAt = time.Now()
 		i.version++
 		return nil
-	case IdentityStatusLocked:
-		return ErrInvalidStatusTransition
-	default:
+	case IdentityStatusLocked, IdentityStatusPendingVerification:
 		return ErrInvalidStatusTransition
 	}
+	return ErrInvalidStatusTransition
 }
 
 // Suspend transitions the identity to SUSPENDED status.
@@ -202,11 +230,10 @@ func (i *Identity) Suspend() error {
 		i.updatedAt = time.Now()
 		i.version++
 		return nil
-	case IdentityStatusPendingInvite, IdentityStatusSuspended, IdentityStatusLocked:
-		return ErrInvalidStatusTransition
-	default:
+	case IdentityStatusPendingInvite, IdentityStatusPendingVerification, IdentityStatusSuspended, IdentityStatusLocked:
 		return ErrInvalidStatusTransition
 	}
+	return ErrInvalidStatusTransition
 }
 
 // Lock transitions the identity to LOCKED status.
@@ -218,11 +245,10 @@ func (i *Identity) Lock() error {
 		i.updatedAt = time.Now()
 		i.version++
 		return nil
-	case IdentityStatusPendingInvite, IdentityStatusLocked:
-		return ErrInvalidStatusTransition
-	default:
+	case IdentityStatusPendingInvite, IdentityStatusPendingVerification, IdentityStatusLocked:
 		return ErrInvalidStatusTransition
 	}
+	return ErrInvalidStatusTransition
 }
 
 // Unlock transitions the identity from LOCKED back to ACTIVE.
@@ -237,6 +263,18 @@ func (i *Identity) Unlock() error {
 	return nil
 }
 
+// Verify transitions the identity from PENDING_VERIFICATION to ACTIVE.
+// Returns ErrNotPendingVerification if the identity is not in PENDING_VERIFICATION status.
+func (i *Identity) Verify() error {
+	if i.status != IdentityStatusPendingVerification {
+		return ErrNotPendingVerification
+	}
+	i.status = IdentityStatusActive
+	i.updatedAt = time.Now()
+	i.version++
+	return nil
+}
+
 // RecordLoginAttempt records a login attempt result. On success it resets the
 // failed attempts counter. On failure it increments the counter and locks the
 // account when the threshold is reached.
@@ -246,12 +284,10 @@ func (i *Identity) RecordLoginAttempt(success bool) error {
 	switch i.status {
 	case IdentityStatusLocked:
 		return ErrAccountLocked
-	case IdentityStatusPendingInvite, IdentityStatusSuspended:
+	case IdentityStatusPendingInvite, IdentityStatusPendingVerification, IdentityStatusSuspended:
 		return ErrInvalidStatusTransition
 	case IdentityStatusActive:
 		// valid — proceed below
-	default:
-		return ErrInvalidStatusTransition
 	}
 
 	if success {

services/identity/domain/identity_test.go

@@ -309,3 +309,70 @@ func TestNewIdentity_EmptyTenantID_ReturnsError(t *testing.T) {
 	assert.Nil(t, identity)
 	assert.ErrorIs(t, err, ErrTenantIDRequired)
 }
+
+func TestNewSelfRegisteredIdentity_VerificationRequired(t *testing.T) {
+	id, err := NewSelfRegisteredIdentity(testTenantID, "self@example.com", true)
+	require.NoError(t, err)
+	assert.NotEqual(t, uuid.Nil, id.ID())
+	assert.Equal(t, "self@example.com", id.Email())
+	assert.Equal(t, IdentityStatusPendingVerification, id.Status())
+	assert.Equal(t, int64(1), id.Version())
+	assert.NotZero(t, id.CreatedAt())
+	assert.NotZero(t, id.UpdatedAt())
+}
+
+func TestNewSelfRegisteredIdentity_VerificationNotRequired(t *testing.T) {
+	id, err := NewSelfRegisteredIdentity(testTenantID, "self@example.com", false)
+	require.NoError(t, err)
+	assert.Equal(t, IdentityStatusActive, id.Status())
+}
+
+func TestNewSelfRegisteredIdentity_InvalidEmail(t *testing.T) {
+	_, err := NewSelfRegisteredIdentity(testTenantID, "not-an-email", true)
+	assert.ErrorIs(t, err, ErrInvalidEmail)
+}
+
+func TestNewSelfRegisteredIdentity_EmptyTenantID(t *testing.T) {
+	_, err := NewSelfRegisteredIdentity("", "self@example.com", true)
+	assert.ErrorIs(t, err, ErrTenantIDRequired)
+}
+
+func TestIdentity_Verify_FromPendingVerification(t *testing.T) {
+	id, err := NewSelfRegisteredIdentity(testTenantID, "self@example.com", true)
+	require.NoError(t, err)
+	require.Equal(t, IdentityStatusPendingVerification, id.Status())
+
+	versionBefore := id.Version()
+	err = id.Verify()
+	require.NoError(t, err)
+	assert.Equal(t, IdentityStatusActive, id.Status())
+	assert.Greater(t, id.Version(), versionBefore)
+}
+
+func TestIdentity_Verify_FromWrongStatus(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup func(*Identity)
+	}{
+		{name: "active", setup: func(id *Identity) { _ = id.Activate() }},
+		{name: "pending invite", setup: func(_ *Identity) {}},
+		{name: "suspended", setup: func(id *Identity) { _ = id.Activate(); _ = id.Suspend() }},
+		{name: "locked", setup: func(id *Identity) { _ = id.Activate(); _ = id.Lock() }},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			id, err := NewIdentity(testTenantID, "user@example.com")
+			require.NoError(t, err)
+			tt.setup(id)
+			err = id.Verify()
+			assert.ErrorIs(t, err, ErrNotPendingVerification)
+		})
+	}
+}
+
+func TestIdentity_RecordLoginAttempt_RejectsPendingVerification(t *testing.T) {
+	id, err := NewSelfRegisteredIdentity(testTenantID, "self@example.com", true)
+	require.NoError(t, err)
+	err = id.RecordLoginAttempt(false)
+	assert.ErrorIs(t, err, ErrInvalidStatusTransition)
+}

services/identity/migrations/20260327000001_drop_identity_status_constraint.sql

@@ -0,0 +1,5 @@
+-- atlas:txn false
+-- Drop the existing identity status check constraint so it can be recreated
+-- with the PENDING_VERIFICATION value in the next migration.
+-- CockroachDB does not allow DROP + ADD of the same constraint name in one transaction.
+ALTER TABLE "identity" DROP CONSTRAINT "chk_identity_status";

services/identity/migrations/20260327000002_add_pending_verification_status.sql

@@ -0,0 +1,7 @@
+-- atlas:txn false
+-- Recreate the identity status check constraint with PENDING_VERIFICATION included.
+-- This runs as a separate migration from the DROP to avoid CockroachDB's restriction
+-- on same-name constraint DROP+ADD within a single transaction.
+ALTER TABLE "identity" ADD CONSTRAINT chk_identity_status CHECK (
+    status IN ('PENDING_INVITE', 'ACTIVE', 'SUSPENDED', 'LOCKED', 'PENDING_VERIFICATION')
+);

services/identity/migrations/atlas.sum

@@ -1,4 +1,6 @@
-h1:s3/ru7NYET/Cs0jNqC0Ndlwp53iY9kthZRxlRIqdizM=
+h1:oWySHCm3QYE9kYHxHDAPc5w0JI2WDeyztYDMP5XKJX8=
 20260302000001_initial.sql h1:SsQN4cGTrm/6qs12GD9YXoSO6QyKTIpNZrM+4rRBWJA=
 20260326000001_add_tenant_id.sql h1:+/NF9LBhT7EjKVqu5gxY1B+vJ+YDM9acG8lLSXGJEWs=
 20260326000002_add_tenant_id_indexes.sql h1:Ov5Wfsl+MTIrtNdnqdh9p5qGpaeBW/wBPXILN4JvuvM=
+20260327000001_drop_identity_status_constraint.sql h1:ctkxpyh2xyGPy1or2dsdWTe0amgbPV0LBMZSbda6Clg=
+20260327000002_add_pending_verification_status.sql h1:U138QbPEfysvyueXoWetj2yVZu2vhHRHLeRa+N8qckc=

services/identity/service/mappers.go

@@ -91,6 +91,8 @@ func domainStatusToProto(s domain.IdentityStatus) pb.IdentityStatus {
 		return pb.IdentityStatus_IDENTITY_STATUS_SUSPENDED
 	case domain.IdentityStatusLocked:
 		return pb.IdentityStatus_IDENTITY_STATUS_LOCKED
+	case domain.IdentityStatusPendingVerification:
+		return pb.IdentityStatus_IDENTITY_STATUS_PENDING_VERIFICATION
 	default:
 		return pb.IdentityStatus_IDENTITY_STATUS_UNSPECIFIED
 	}

services/identity/service/mappers_test.go

@@ -39,6 +39,16 @@ func TestIdentityToProto_PendingInviteStatus(t *testing.T) {
 	assert.Equal(t, pb.IdentityStatus_IDENTITY_STATUS_PENDING_INVITE, proto.Status)
 }
 
+func TestIdentityToProto_PendingVerificationStatus(t *testing.T) {
+	identity, err := domain.NewSelfRegisteredIdentity(mapperTestTID, "self@example.com", true)
+	require.NoError(t, err)
+
+	proto := identityToProto(identity)
+
+	require.NotNil(t, proto)
+	assert.Equal(t, pb.IdentityStatus_IDENTITY_STATUS_PENDING_VERIFICATION, proto.Status)
+}
+
 func TestIdentityToProto_TimestampsPopulated(t *testing.T) {
 	before := time.Now().Truncate(time.Second)
 	identity, err := domain.NewIdentity(mapperTestTID, "ts@example.com")