feat: scope candidate profile uniqueness

docs/phase-2-architecture.md

@@ -48,7 +48,7 @@ Stage 1 components perform no external calls. The Stage 2 endpoint performs only
 
 ## Flyway migration
 
-`V2__phase_2_persistence_and_candidate_profile.sql` is the schema source of truth. Hibernate remains configured with `ddl-auto: validate`.
+Flyway migrations are the schema source of truth. Hibernate remains configured with `ddl-auto: validate`.
 
 The migration adds:
 
@@ -63,8 +63,8 @@ The migration adds:
 
 Important database guarantees include:
 
-- a positive, unique profile version;
-- exactly zero or one active profile, enforced with a unique nullable `active_slot` plus a consistency check;
+- a positive profile version that is unique within its candidate;
+- exactly zero or one active profile per candidate, enforced with candidate-scoped uniqueness on the nullable `active_slot` plus a consistency check;
 - stable-key uniqueness in each candidate fact scope;
 - normalized skill, language, and project uniqueness within a profile version;
 - one application per job;
@@ -81,7 +81,7 @@ Startup invokes a small runner which delegates to `CandidateProfileBootstrapServ
 
 1. Validate Bean Validation constraints and cross-record truth rules.
 2. Compute a SHA-256 source fingerprint without logging the source.
-3. Compare the configured version with the active database version.
+3. Compare the configured version with that candidate's active database version.
 4. Return the existing row when both version and fingerprint match.
 5. Reject a lower version or changed facts using the same version.
 6. Deactivate the current row and insert a complete higher version atomically.

src/main/java/com/jobpilot/candidate/domain/CandidateProfile.java

@@ -28,11 +28,7 @@ public class CandidateProfile {
     @ManyToOne(fetch = FetchType.LAZY, optional = false)
     @JoinColumn(name = "candidate_id", nullable = false)
     private Candidate candidate;
-    /**
-     * Still globally unique in the schema, not unique per candidate. Making version and active
-     * uniqueness candidate-scoped is a separate migration.
-     */
-    @Column(nullable = false, unique = true)
+    @Column(nullable = false)
     private int profileVersion;
     @Column(nullable = false, length = 200)
     private String fullName;

src/main/java/com/jobpilot/candidate/repository/CandidateProfileRepository.java

@@ -10,10 +10,10 @@ public interface CandidateProfileRepository extends JpaRepository<CandidateProfi
     Optional<CandidateProfile> findByCandidateIdAndProfileVersion(Long candidateId, int profileVersion);
 
     /**
-     * Installation-wide lookups, meaning "the one active profile in the whole system". Still
-     * correct only because a single candidate exists; {@code JobAnalysisService} and
-     * {@code ResumeGenerationService} read the active profile this way and are migrated to
-     * candidate-scoped access in a later phase.
+     * Legacy installation-wide lookups. The database permits version reuse and one active profile
+     * per candidate, so these methods are safe only while their callers operate with one configured
+     * candidate. {@code JobAnalysisService} and {@code ResumeGenerationService} are migrated to
+     * candidate-scoped access when candidate context reaches those workflows.
      */
     Optional<CandidateProfile> findByActiveTrue();
 

src/main/java/db/migration/V15__candidate_profile_scoped_uniqueness.java

@@ -0,0 +1,124 @@
+package db.migration;
+
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.zip.CRC32;
+import org.flywaydb.core.api.migration.BaseJavaMigration;
+import org.flywaydb.core.api.migration.Context;
+
+public class V15__candidate_profile_scoped_uniqueness extends BaseJavaMigration {
+    private static final String TABLE = "candidate_profiles";
+    private static final String CANDIDATE_ID = "candidate_id";
+    private static final String PROFILE_VERSION = "profile_version";
+    private static final String ACTIVE_SLOT = "active_slot";
+    private static final String PROFILE_VERSION_CONSTRAINT =
+            "candidate_profiles_candidate_profile_version_uk";
+    private static final String ACTIVE_SLOT_CONSTRAINT =
+            "candidate_profiles_candidate_active_slot_uk";
+    private static final String MIGRATION_DEFINITION = String.join("\n",
+            "table=resolved-schema." + TABLE,
+            "discover=single-column-unique:" + PROFILE_VERSION,
+            "discover=single-column-unique:" + ACTIVE_SLOT,
+            "add=" + PROFILE_VERSION_CONSTRAINT + ":" + CANDIDATE_ID + "," + PROFILE_VERSION,
+            "add=" + ACTIVE_SLOT_CONSTRAINT + ":" + CANDIDATE_ID + "," + ACTIVE_SLOT,
+            "drop=discovered-global:" + PROFILE_VERSION,
+            "drop=discovered-global:" + ACTIVE_SLOT);
+
+    /** Java migrations have no checksum by default, so hash the canonical schema definition. */
+    @Override
+    public Integer getChecksum() {
+        CRC32 checksum = new CRC32();
+        checksum.update(MIGRATION_DEFINITION.getBytes(StandardCharsets.UTF_8));
+        return (int) checksum.getValue();
+    }
+
+    @Override
+    public void migrate(Context context) throws Exception {
+        Connection connection = context.getConnection();
+        String schema = connection.getSchema();
+        if (schema == null || schema.isBlank()) {
+            throw new IllegalStateException("Candidate profile migration requires a current schema");
+        }
+        Map<String, List<String>> uniqueConstraints = uniqueConstraints(connection, schema);
+        String profileVersionConstraint = singleColumnConstraint(
+                uniqueConstraints, PROFILE_VERSION);
+        String activeSlotConstraint = singleColumnConstraint(uniqueConstraints, ACTIVE_SLOT);
+        if (profileVersionConstraint.equals(activeSlotConstraint)) {
+            throw new IllegalStateException(
+                    "Candidate profile global UNIQUE constraints must have distinct names");
+        }
+
+        String table = quote(connection, schema) + "." + quote(connection, TABLE);
+        try (var statement = connection.createStatement()) {
+            statement.execute("ALTER TABLE " + table + " ADD CONSTRAINT "
+                    + quote(connection, PROFILE_VERSION_CONSTRAINT)
+                    + " UNIQUE (" + quote(connection, CANDIDATE_ID) + ", "
+                    + quote(connection, PROFILE_VERSION) + ")");
+            statement.execute("ALTER TABLE " + table + " ADD CONSTRAINT "
+                    + quote(connection, ACTIVE_SLOT_CONSTRAINT)
+                    + " UNIQUE (" + quote(connection, CANDIDATE_ID) + ", "
+                    + quote(connection, ACTIVE_SLOT) + ")");
+            statement.execute("ALTER TABLE " + table + " DROP CONSTRAINT "
+                    + quote(connection, profileVersionConstraint));
+            statement.execute("ALTER TABLE " + table + " DROP CONSTRAINT "
+                    + quote(connection, activeSlotConstraint));
+        }
+    }
+
+    private Map<String, List<String>> uniqueConstraints(Connection connection, String schema)
+            throws SQLException {
+        String sql = """
+                SELECT tc.constraint_name, kcu.column_name
+                FROM information_schema.table_constraints tc
+                JOIN information_schema.key_column_usage kcu
+                  ON kcu.constraint_catalog = tc.constraint_catalog
+                 AND kcu.constraint_schema = tc.constraint_schema
+                 AND kcu.constraint_name = tc.constraint_name
+                 AND kcu.table_catalog = tc.table_catalog
+                 AND kcu.table_schema = tc.table_schema
+                 AND kcu.table_name = tc.table_name
+                WHERE tc.table_schema = ?
+                  AND tc.table_name = ?
+                  AND tc.constraint_type = 'UNIQUE'
+                ORDER BY tc.constraint_name, kcu.ordinal_position
+                """;
+        Map<String, List<String>> constraints = new LinkedHashMap<>();
+        try (var statement = connection.prepareStatement(sql)) {
+            statement.setString(1, schema);
+            statement.setString(2, TABLE);
+            try (ResultSet rows = statement.executeQuery()) {
+                while (rows.next()) {
+                    constraints.computeIfAbsent(rows.getString(1), ignored -> new ArrayList<>())
+                            .add(rows.getString(2));
+                }
+            }
+        }
+        return constraints;
+    }
+
+    private String singleColumnConstraint(Map<String, List<String>> constraints, String column) {
+        List<String> matches = constraints.entrySet().stream()
+                .filter(entry -> entry.getValue().equals(List.of(column)))
+                .map(Map.Entry::getKey)
+                .toList();
+        if (matches.size() != 1) {
+            throw new IllegalStateException("Expected exactly one single-column UNIQUE constraint on "
+                    + TABLE + "(" + column + "), found " + matches.size() + ": " + matches);
+        }
+        return matches.getFirst();
+    }
+
+    private String quote(Connection connection, String identifier) throws SQLException {
+        String quote = connection.getMetaData().getIdentifierQuoteString();
+        if (quote == null || quote.isBlank()) {
+            throw new IllegalStateException("Database does not provide an identifier quote string");
+        }
+        return quote + identifier.replace(quote, quote + quote) + quote;
+    }
+}

src/test/java/com/jobpilot/candidate/V14CandidateOwnershipMigrationTest.java

@@ -28,7 +28,7 @@ void backfillsExistingProfilesToTheSeededDefaultCandidate() throws Exception {
             statement.executeUpdate(profile(2, true));
         }
 
-        Flyway.configure().dataSource(url, "sa", "").load().migrate();
+        Flyway.configure().dataSource(url, "sa", "").target("14").load().migrate();
         try (Connection connection = DriverManager.getConnection(url, "sa", "");
              Statement statement = connection.createStatement()) {
             try (var rows = statement.executeQuery(
@@ -136,7 +136,7 @@ void profileVersionAndActiveSlotRemainGloballyUnique() throws Exception {
 
     private String migrate(String name) {
         String url = "jdbc:h2:mem:" + name + ";MODE=PostgreSQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE";
-        Flyway.configure().dataSource(url, "sa", "").load().migrate();
+        Flyway.configure().dataSource(url, "sa", "").target("14").load().migrate();
         return url;
     }
 

src/test/java/com/jobpilot/candidate/V15CandidateProfileScopedUniquenessMigrationTest.java

@@ -0,0 +1,188 @@
+package com.jobpilot.candidate;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import db.migration.V15__candidate_profile_scoped_uniqueness;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.flywaydb.core.Flyway;
+import org.junit.jupiter.api.Test;
+
+class V15CandidateProfileScopedUniquenessMigrationTest {
+    private static final String VERSION_CONSTRAINT =
+            "candidate_profiles_candidate_profile_version_uk";
+    private static final String ACTIVE_CONSTRAINT =
+            "candidate_profiles_candidate_active_slot_uk";
+
+    @Test
+    void upgradesV14DataAndScopesVersionAndActiveUniquenessByCandidate() throws Exception {
+        String url = url("v15upgrade");
+        migrate(url, "14");
+        Map<Integer, StoredProfile> before;
+        long owner;
+        try (Connection connection = DriverManager.getConnection(url, "sa", "");
+             Statement statement = connection.createStatement()) {
+            owner = candidateId(statement, "default");
+            statement.executeUpdate(ownedProfile(1, false, owner));
+            statement.executeUpdate(ownedProfile(2, true, owner));
+            before = profiles(statement);
+        }
+
+        migrate(url, "15");
+        try (Connection connection = DriverManager.getConnection(url, "sa", "");
+             Statement statement = connection.createStatement()) {
+            assertThat(profiles(statement)).isEqualTo(before);
+
+            statement.executeUpdate(candidate("second-candidate"));
+            long other = candidateId(statement, "second-candidate");
+            statement.executeUpdate(ownedProfile(1, true, other));
+
+            assertThatThrownBy(() -> statement.executeUpdate(ownedProfile(1, false, other)))
+                    .as("duplicate profile version for one candidate")
+                    .isInstanceOf(SQLException.class);
+            assertThatThrownBy(() -> statement.executeUpdate(ownedProfile(2, true, other)))
+                    .as("second active profile for one candidate")
+                    .isInstanceOf(SQLException.class);
+            assertThatThrownBy(() -> statement.executeUpdate(profile(3, false)))
+                    .as("missing candidate_id")
+                    .isInstanceOf(SQLException.class);
+            assertThatThrownBy(() -> statement.executeUpdate(ownedProfile(3, false, 987_654L)))
+                    .as("unknown candidate_id")
+                    .isInstanceOf(SQLException.class);
+            assertThatThrownBy(() -> statement.executeUpdate(
+                    "DELETE FROM candidates WHERE id = " + other))
+                    .as("candidate owning profiles remains protected by ON DELETE RESTRICT")
+                    .isInstanceOf(SQLException.class);
+
+            Map<String, List<String>> constraints = uniqueConstraints(statement);
+            assertThat(constraints)
+                    .containsEntry(VERSION_CONSTRAINT, List.of("candidate_id", "profile_version"))
+                    .containsEntry(ACTIVE_CONSTRAINT, List.of("candidate_id", "active_slot"));
+            assertThat(constraints.values())
+                    .noneMatch(columns -> columns.equals(List.of("profile_version")))
+                    .noneMatch(columns -> columns.equals(List.of("active_slot")));
+        }
+    }
+
+    @Test
+    void migratesAFreshDatabaseThroughV15() throws Exception {
+        String url = url("v15fresh");
+        migrate(url, "15");
+
+        try (Connection connection = DriverManager.getConnection(url, "sa", "");
+             Statement statement = connection.createStatement()) {
+            try (var rows = statement.executeQuery(
+                    "SELECT COUNT(*) c, MAX(checksum) checksum FROM flyway_schema_history "
+                            + "WHERE version = '15' AND success = TRUE")) {
+                assertThat(rows.next()).isTrue();
+                assertThat(rows.getInt("c")).isOne();
+                assertThat(rows.getInt("checksum")).isEqualTo(
+                        new V15__candidate_profile_scoped_uniqueness().getChecksum());
+            }
+            try (var rows = statement.executeQuery("SELECT COUNT(*) c FROM candidate_profiles")) {
+                assertThat(rows.next()).isTrue();
+                assertThat(rows.getInt("c")).isZero();
+            }
+            assertThat(uniqueConstraints(statement))
+                    .containsEntry(VERSION_CONSTRAINT, List.of("candidate_id", "profile_version"))
+                    .containsEntry(ACTIVE_CONSTRAINT, List.of("candidate_id", "active_slot"));
+        }
+    }
+
+    private void migrate(String url, String target) {
+        Flyway.configure().dataSource(url, "sa", "").target(target).load().migrate();
+    }
+
+    private String url(String name) {
+        return "jdbc:h2:mem:" + name
+                + ";MODE=PostgreSQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE";
+    }
+
+    private Map<Integer, StoredProfile> profiles(Statement statement) throws Exception {
+        Map<Integer, StoredProfile> profiles = new LinkedHashMap<>();
+        try (var rows = statement.executeQuery("""
+                SELECT id, candidate_id, profile_version, active, active_slot
+                FROM candidate_profiles ORDER BY profile_version
+                """)) {
+            while (rows.next()) {
+                short activeSlotValue = rows.getShort("active_slot");
+                Short activeSlot = rows.wasNull() ? null : activeSlotValue;
+                profiles.put(rows.getInt("profile_version"), new StoredProfile(
+                        rows.getLong("id"), rows.getLong("candidate_id"),
+                        rows.getInt("profile_version"), rows.getBoolean("active"),
+                        activeSlot));
+            }
+        }
+        return profiles;
+    }
+
+    private Map<String, List<String>> uniqueConstraints(Statement statement) throws Exception {
+        Map<String, List<String>> constraints = new LinkedHashMap<>();
+        try (var rows = statement.executeQuery("""
+                SELECT tc.constraint_name, kcu.column_name
+                FROM information_schema.table_constraints tc
+                JOIN information_schema.key_column_usage kcu
+                  ON kcu.constraint_catalog = tc.constraint_catalog
+                 AND kcu.constraint_schema = tc.constraint_schema
+                 AND kcu.constraint_name = tc.constraint_name
+                 AND kcu.table_catalog = tc.table_catalog
+                 AND kcu.table_schema = tc.table_schema
+                 AND kcu.table_name = tc.table_name
+                WHERE tc.table_name = 'candidate_profiles'
+                  AND tc.constraint_type = 'UNIQUE'
+                ORDER BY tc.constraint_name, kcu.ordinal_position
+                """)) {
+            while (rows.next()) {
+                constraints.computeIfAbsent(rows.getString("constraint_name"),
+                                ignored -> new java.util.ArrayList<>())
+                        .add(rows.getString("column_name"));
+            }
+        }
+        return constraints;
+    }
+
+    private long candidateId(Statement statement, String stableKey) throws Exception {
+        try (var rows = statement.executeQuery(
+                "SELECT id FROM candidates WHERE stable_key = '" + stableKey + "'")) {
+            assertThat(rows.next()).isTrue();
+            return rows.getLong("id");
+        }
+    }
+
+    private String candidate(String stableKey) {
+        return "INSERT INTO candidates (stable_key, created_at, updated_at) VALUES ('"
+                + stableKey + "', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)";
+    }
+
+    private String profile(int version, boolean active) {
+        return profileColumns("", version, active, "");
+    }
+
+    private String ownedProfile(int version, boolean active, long candidateId) {
+        return profileColumns("candidate_id, ", version, active, candidateId + ", ");
+    }
+
+    private String profileColumns(String ownerColumn, int version, boolean active,
+                                  String ownerValue) {
+        return """
+                INSERT INTO candidate_profiles (%sprofile_version, full_name, location,
+                    education_institution, degree, study_start_year, current_student,
+                    final_year_student, commercial_java_experience_years, source_hash,
+                    created_at, updated_at, active, active_slot)
+                VALUES (%s%d, 'Stored Candidate', 'Bucharest, Romania', 'Stored University',
+                    'BSc', 2025, TRUE, FALSE, 0, 'hash-%d', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP,
+                    %s, %s)
+                """.formatted(ownerColumn, ownerValue, version, version,
+                active ? "TRUE" : "FALSE", active ? "1" : "NULL");
+    }
+
+    private record StoredProfile(long id, long candidateId, int profileVersion,
+                                 boolean active, Short activeSlot) {
+    }
+}