객체 54 개 중
무엇이 실제로 열리나

// 세 규칙이 CI 에서 돈다. 루트 패키지 이름은 구현 저장소가 정한다.
@AnalyzeClasses(packages = ROOT_PACKAGE, importOptions = ImportOption.DoNotIncludeTests.class)
class PackageBoundaryTest {

    @ArchTest
    static final ArchRule collector_는_processor_와_api_를_모른다 =
            noClasses().that().resideInAPackage("..collector..")
                    .should().dependOnClassesThat().resideInAnyPackage("..processor..", "..api..");

    @ArchTest
    static final ArchRule processor_는_collector_와_api_를_모른다 =
            noClasses().that().resideInAPackage("..processor..")
                    .should().dependOnClassesThat().resideInAnyPackage("..collector..", "..api..");

    @ArchTest
    static final ArchRule api_는_읽기만_한다 =
            noClasses().that().resideInAPackage("..api..")
                    .should().dependOnClassesThat().resideInAnyPackage("..collector..", "..processor..");
}

/**
 * processor 가 읽는 관측 한 행. vehicle_observation 한 행이고 열 이름을 그대로 쓴다.
 * collector 의 VehicleObservation 과 형을 공유하지 않는다 — ArchUnit 이 그 의존을 막는다.
 */
public record ObservedVehicle(
        long id,                    // vehicle_stop_prediction.vehicleObservationId 가 이 값이다
        long routeReferenceVersionId,
        String vehicleId,           // 상류가 빠뜨리면 null. 응답의 StopArrival.vehicleId 가 이 값이다
        String vehicleTripKey,      // v4 신규 열. null 이어도 예보는 낸다 — 라벨 회수만 못 한다
        int stopOrder,              // 지금 통과·접근 중인 순번. 예보 대상은 +1 부터 +12 까지다
        Instant observedAt,
        Integer remainingSeats,     // null = 좌석 결측. 0 = 만석(입석이 없어 승차 불가와 같다)
        Integer crowdedCode,        // 1~4 만 유효. 원문 0 은 미산출이라 null
        Integer lowPlateCode        // 2 = 2층버스. 정원은 이 값에서 유도하지 않는다
) {}

/**
 * 예보 하나를 내는 데 필요한 입력 전부. v3 ForecastTarget 을 대신한다.
 *
 * 지평은 targetStopOrder 에서 유도되는 값인데도 따로 들고 다닌다 — 모델 키라서
 * 계수 조회와 채점이 이 축을 그대로 쓰기 때문이다.
 */
public record VehicleStopTarget(ObservedVehicle observation, int targetStopOrder, Horizon horizon) {

    public VehicleStopTarget {
        Objects.requireNonNull(observation, "observation");
        Objects.requireNonNull(horizon, "horizon");
        if (targetStopOrder != observation.stopOrder() + horizon.stops()) {
            throw new IllegalArgumentException("지평과 순번 차가 어긋난다: target=" + targetStopOrder
                    + " observed=" + observation.stopOrder() + " horizon=" + horizon.stops());
        }
        if (observation.remainingSeats() == null) {
            // 설계행렬이 잔여석을 그대로 쓰고 앵커도 거기서 출발하는데, 그 자리에는
            // 상류 기울기·직전 좌석과 달리 결측 지시자가 없다. 대상 자체를 만들지 않는다.
            throw new IllegalArgumentException("좌석 결측 관측으로는 대상을 만들 수 없다: obs=" + observation.id());
        }
    }

    /** 예측 시점의 그 차량 잔여석. v3 입력에 없던 값이고 A18 의 핵심 재료다. */
    public int remainingSeats() {
        return observation.remainingSeats();
    }
}

/**
 * 도착 시 잔여좌석의 분포. 좌석 수마다 확률을 하나씩 담는다.
 * 분포 전체를 DB 에 저장하지는 않는다 — 사이트 평가 규약의 지표가 전부 만석 여부 이진 사건에
 * 대한 것이라 pFull 한 열이면 사후 채점이 된다.
 */
public record SeatDistribution(double[] mass) {

    private static final double SUM_TOLERANCE = 1e-9;

    public SeatDistribution {
        Objects.requireNonNull(mass, "mass");
        if (mass.length == 0) {
            throw new IllegalArgumentException("분포가 비어 있다");
        }
        mass = mass.clone();   // record 라도 배열은 복사해야 불변이 된다
        double sum = 0;
        for (int k = 0; k < mass.length; k++) {
            double p = mass[k];
            if (!Double.isFinite(p)) {                     // NaN · ±Infinity 는 ck_vsp_pfull 도 거절한다
                throw new IllegalArgumentException("확률이 유한하지 않다: k=" + k);
            }
            if (p < 0 || p > 1) {
                throw new IllegalArgumentException("확률이 0~1 밖이다: k=" + k + " p=" + p);
            }
            sum += p;
        }
        if (Math.abs(sum - 1.0) > SUM_TOLERANCE) {
            throw new IllegalArgumentException("확률 합이 1 이 아니다: " + sum);
        }
    }

    /**
     * 정원. 그 차량이 보여 준 최대 잔여석이고 형식승인 좌석수와는 다른 양이다.
     * DB 열이 아니라 번들의 특징 계약이 소유한다 — 종전 47석 고정은 어느 근거에도 없던 수였다.
     */
    public int capacity() {
        return mass.length - 1;
    }

    /**
     * 만석 확률. 분포의 0 지점이다. vehicle_stop_prediction.pFull 이 이 값이다.
     * 반올림하지 않는다 — 0.996 이 1.00 이 되면 "반드시 자리가 있다"로 읽힌다.
     */
    public double pFull() {
        return mass[0];
    }

    /**
     * 잔여좌석 기댓값. vehicle_stop_prediction.expectedSeats 가 이 값이다.
     * 분포를 요약한 값이므로 이 값이 없어도 pFull 은 유효하다.
     */
    public double expectedSeats() {
        double e = 0;
        for (int k = 1; k < mass.length; k++) {
            e += k * mass[k];
        }
        return e;
    }

    // 빈자리 확률(1 − pFull)은 여기서 내지 않는다. 뒤집기는 api 서비스 계층 메서드 한 곳에서만 하고
    // SQL 안에서도 하지 않는다 — 두 곳에서 뒤집으면 어느 쪽이 뒤집은 값인지 모르게 된다.

    @Override
    public double[] mass() {
        return mass.clone();
    }

    @Override
    public boolean equals(Object o) {                       // 배열이라 record 기본 구현을 쓸 수 없다
        return o instanceof SeatDistribution other && Arrays.equals(mass, other.mass);
    }

    @Override
    public int hashCode() {
        return Arrays.hashCode(mass);
    }
}

/**
 * 몇 정류장 전에서 예측했는가. 시간이 아니라 정류장 수다.
 *
 * 같은 값이 세 곳에 박힌다 — 이 상수 · vehicle_stop_prediction.ck_vsp_horizon ·
 * 두 계약의 horizonStops maximum. 상한을 바꾸려면 셋을 함께 연다(4절).
 */
public record Horizon(int stops) {

    public static final int MIN = 1;
    public static final int MAX = 12;

    public Horizon {
        if (stops < MIN || stops > MAX) {
            throw new IllegalArgumentException("지평은 " + MIN + "~" + MAX + " 다: " + stops);
        }
    }

    public static Horizon of(int stops) {
        return new Horizon(stops);
    }

    /** 관측 하나가 낼 수 있는 지평 전부. 대상이 여정 끝을 넘으면 호출 쪽에서 끊는다. */
    public static List<Horizon> all() {
        return IntStream.rangeClosed(MIN, MAX).mapToObj(Horizon::new).toList();
    }
}

/**
 * 좌석 예보 모델 계약. 서버는 이 계약을 실행만 한다 — 학습하지 않는다.
 * 지원 범위 밖이면 절편만으로 조용히 값을 내지 않는다. 그 판정은 승인 참조가 갖는다.
 */
public interface SeatForecastModel {

    /** 이 구현을 실은 승인 release 참조. covers() 가 노선 판본과 지평 1~12 를 함께 본다. */
    ApprovedModelRef approvedRef();

    /**
     * 대상 하나에 도착 시 좌석 분포를 낸다.
     * profile 은 그 노선 판본 · 그 시간대의 셀 전부다 — 셀 하나가 아니다.
     */
    SeatForecastResult predict(VehicleStopTarget target, DemandProfile profile);
}

/** 예보 하나의 결과. 열이 둘이라 값도 둘이다. */
public record SeatForecastResult(SeatDistribution distribution, double pFullRaw) {

    public SeatForecastResult {
        Objects.requireNonNull(distribution, "distribution");
        if (!Double.isFinite(pFullRaw) || pFullRaw < 0 || pFullRaw > 1) {
            throw new IllegalArgumentException("pFullRaw 가 0~1 밖이다: " + pFullRaw);
        }
    }

    /** 응답과 채점에 쓰는 만석 확률. 분포의 0 지점이다. */
    public double pFull() {
        return distribution.pFull();
    }
}

/**
 * A18(차량 상태 조건부 좌석 분포)의 서빙 구현.
 * 계수는 외부 ML 환경이 승인한 번들에서 오고 이 클래스는 그것을 실행만 한다.
 */
public final class SeatDistributionModel implements SeatForecastModel {

    private final ApprovedModelRef approvedRef;
    private final CoefficientBundle bundle;   // 노선별 블록 · 지평 1~12 계수 전부 · golden vector 로 검증된 것

    public SeatDistributionModel(ApprovedModelRef approvedRef, CoefficientBundle bundle) {
        this.approvedRef = Objects.requireNonNull(approvedRef);
        this.bundle = Objects.requireNonNull(bundle);
    }

    @Override
    public ApprovedModelRef approvedRef() {
        return approvedRef;
    }

    /**
     * 층을 통과시켜 분포를 만들고 마지막에 사전확률 이동을 얹는다.
     * 이동 전 만석 확률이 pFullRaw 이고 얹은 뒤가 pFull 이다.
     * 층 정의와 수식은 A18 모델 문서에 있다 — 여기 옮기지 않는다.
     */
    @Override
    public SeatForecastResult predict(VehicleStopTarget target, DemandProfile profile) {
        /* 골격 — 본체는 A18 문서의 층 정의를 따른다 */
    }
}

/**
 * 같은 차량의 한 여정을 잇는 키. vehicle_observation.vehicleTripKey 가 이 값의 텍스트다.
 * 여정 시작 시각은 같은 키를 가진 행들의 최소 observedAt 이므로 열을 따로 두지 않는다.
 */
public record VehicleTripKey(String sourceId, String vehicleId,
                             long routeReferenceVersionId, Instant startedAt) {

    private static final int COLUMN_MAX_LENGTH = 120;   // varchar(120)

    public VehicleTripKey {
        Objects.requireNonNull(sourceId, "sourceId");
        Objects.requireNonNull(startedAt, "startedAt");
        // vehicleId 가 없으면 여정도 특정할 수 없다. 그때는 이 객체를 만들지 않고 열에 null 을 넣는다 —
        // CHECK 가 "vehicleId 가 null 이면 vehicleTripKey 도 null" 을 강제한다.
        Objects.requireNonNull(vehicleId, "vehicleId");
    }

    /** 열에 저장하는 텍스트. 전역 ID 가 아니라 (sourceId, vehicleId) 가 키다. */
    public String asColumn() {
        String value = sourceId + '|' + vehicleId + '|' + routeReferenceVersionId + '|' + startedAt;
        if (value.length() > COLUMN_MAX_LENGTH) {
            throw new IllegalStateException("여정 키가 열 길이를 넘는다: " + value.length());
        }
        return value;
    }

    /**
     * 직전 관측과 견줘 새 여정이 시작됐는지 본다. 셋 중 하나면 새 여정이고 그 밖에는 직전 키를 잇는다.
     *
     *   1. stopOrder 가 감소한다            — 문턱 미결. 상류 지터로 잠깐 되돌아가는 경우와 구별해야 한다
     *   2. 직전 관측과의 간격이 문턱을 넘는다 — 문턱 미결. 회차 대기가 시간대에 따라 변한다
     *                                        (1650 기준 중앙이 07시 106.7분 → 21시 46.0분)
     *   3. routeReferenceVersionId 또는 poll 의 normalizationVersion 이 다르다
     *
     * 3330 은 회차 순번 언저리(42~43)에서 끝나고 다시 시작하는 여정이 있어 순번만으로는 갈리지 않는다.
     */
    public static boolean startsNewTrip(VehicleObservation previous, VehicleObservation next) {
        /* 골격 — 문턱 둘이 미결이라 값을 아직 못 박는다 */
    }
}

/** KST 시간대 3구분. v3 그대로 쓴다. demand_profile.timeCellId 가 이 값이다. */
public enum TimeBand {
    MORNING,   // 07~09
    EVENING,   // 17~20
    OTHER;

    /** 열에 저장하는 문자열. 셀 정의의 소유자는 이 enum 이 아니라 featureContractVersion 이다. */
    public String timeCellId() {
        return name().toLowerCase(Locale.ROOT);
    }

    /**
     * 서빙은 예측 시각의 시간대를 쓴다 — 도착 시각을 모르기 때문이다.
     * 학습 라벨의 시간대는 도착 관측 시각에서 왔고, 지평 12 의 중앙 소요가 25.0분이라
     * 경계를 넘는 경우가 생긴다(미결 — 시간대 축).
     */
    public static TimeBand of(Instant at) {
        /* 골격 — KST 로 옮겨 07~09 · 17~20 을 가른다 */
    }
}

/** 셀 하나. demand_profile 한 행이고 열 이름을 그대로 쓴다. */
public record DemandProfileCell(
        int stopOrder,
        double occupancyMean,   // 그 정류장 도착 시 평균 점유율. 날짜 균등가중이다
        double netDemandMean,   // 평균 순수요 ÷ 정원. 음수가 정상이다 — 하차 우세 정류장이 그렇다
        int sampleCount,        // 이 셀을 만든 유효 라벨 수. 이웃 폴백 판정의 재료다
        int dayCount,           // 날짜 균등가중이라 sampleCount 만으로는 평균을 다시 합칠 수 없다
        Instant trainedThrough  // 이 시각까지의 도착 라벨만 들어갔다. 누출 차단 기준이다
) {}

/**
 * 한 노선 판본 · 한 시간대의 셀 전부. 정류장 수만큼 들어 있다.
 * z화 · 이웃 폴백 · 구간합이 전부 이 객체 안에서 닫힌다.
 */
public record DemandProfile(
        long routeReferenceVersionId,
        TimeBand timeBand,
        String featureContractVersion,
        int revision,                                     // 재계산 세대. cellProfileRevision 에 그대로 적는다
        NavigableMap<Integer, DemandProfileCell> cells    // key = stopOrder
) {
    /** 이웃 폴백 반경. 1/d² 가중이고 자기 자신은 뺀다. */
    public static final int FALLBACK_RADIUS = 4;

    public DemandProfile {
        Objects.requireNonNull(timeBand, "timeBand");
        Objects.requireNonNull(featureContractVersion, "featureContractVersion");
        if (revision < 1) {
            throw new IllegalArgumentException("세대 번호는 1 부터다: " + revision);
        }
        cells = Collections.unmodifiableNavigableMap(new TreeMap<>(cells));
    }

    /** 폴백 지시자를 열로 두지 않는 이유 — 그 셀의 행이 있는지가 곧 지시자다. */
    public boolean hasCell(int stopOrder) {
        return cells.containsKey(stopOrder);
    }

    /** 점유율의 z 값. 표준화 상수를 저장하지 않고 이 객체가 든 행들에서 그때 유도한다. */
    public double occupancyZ(int stopOrder) {
        /* 골격 — 규칙은 A18 모델 문서 */
    }

    /** 순수요의 z 값. 셀이 없으면 반경 4 안의 이웃이 받는다. */
    public double netDemandZ(int stopOrder) {
        /* 골격 — 규칙은 A18 모델 문서 */
    }

    /** 관측 순번 다음부터 대상 순번까지의 구간합. 지평이 길수록 이 값이 예보를 떠받친다. */
    public double netDemandSegmentSum(int fromExclusive, int toInclusive) {
        /* 골격 — 규칙은 A18 모델 문서 */
    }
}

/**
 * 셀 통계 읽기 경계. 메서드가 하나뿐인 것이 이 포트의 요점이다.
 *
 * 셀 하나를 집어 오는 메서드를 두면 z화가 깨진다 — 표준화 상수를 저장하지 않기 때문이다.
 * 이웃 폴백(반경 4)도 구간합도 이웃 행이 손에 있어야 닫힌다.
 * featureContractVersion 이 키에 들어가는 이유는 정원이 바뀌면 값 자체가 달라져서다 —
 * 옛 정원으로 계산한 행과 새 정원으로 계산한 행이 섞이지 않는다.
 */
public interface DemandProfileRepository {

    DemandProfile load(long routeReferenceVersionId, TimeBand timeBand, String featureContractVersion);
}

/** 라벨 회수 상태. vehicle_stop_prediction.settlement 가 이 값이다. */
public enum Settlement {

    PENDING,        // 아직 도착 전이거나 회수 배치가 안 돌았다
    SETTLED,        // 같은 여정 안에서 대상 순번의 관측을 찾았고 잔여석이 유효하다. arrivedSeats NOT NULL
    SEAT_MISSING,   // 도착 관측은 찾았는데 좌석이 결측이었다. 관측 FK 는 있고 arrivedSeats 는 null
    SKIPPED,        // 여정이 대상 순번을 건너뛰었다 — v3 라벨 조건이 학습에서 빼는 그 경우다
    LOST;           // 대상 순번에 닿기 전에 여정이 끊겼다. 여정 키가 없는 행도 여기로 닫는다

    /** 채점 · demand_profile · 사전확률 이동은 SETTLED 행만 쓴다. */
    public boolean scorable() {
        return this == SETTLED;
    }
}

/** 라벨 회수 배치. PENDING 행을 나중 관측과 맞춰 닫는다. */
public interface SettleArrivalLabelsJob {

    /** 한 번 돈다. 회수 주기와 창은 미결이다. */
    SettlementReport run(Instant now);
}

/** 한 번 돈 결과. 넷을 갈라 세는 이유는 왜 못 썼는지가 남아야 하기 때문이다. */
public record SettlementReport(int settled, int seatMissing, int skipped, int lost, int stillPending) {}

/**
 * 예보 쓰기와 라벨 회수의 저장 경계.
 * 한 poll 의 행 쓰기와 완료 표시를 같은 포트에 둔 것은 둘이 같은 transaction 안에서 끝나야 하기 때문이다.
 */
public interface VehicleStopPredictionRepository {

    /** 한 poll 의 행 전부. 한 transaction · 한 배포 · 같은 generatedAt 이다. */
    void saveAll(List<VehicleStopPrediction> rows);

    /** 다 쓰면 찍는다. 예보 대상이 0건이어도 찍는다 — 안 찍으면 그 판이 스냅샷 후보에서 빠진다. */
    void markForecastCompleted(long pollId, Instant completedAt);

    /** 회수 대상. settlement = PENDING 이고 대상 순번을 지났을 법한 행이다. */
    List<PendingLabel> findPending(long routeReferenceVersionId, Instant before, int limit);

    /** 회수 결과를 한 행에 적는다. 상태와 세 값의 조합은 DB CHECK 가 다시 막는다. */
    void settle(long vehicleObservationId, int targetStopOrder, Settlement settlement,
                Long outcomeObservationId, Integer arrivedSeats, Instant settledAt);
}

/**
 * v4 의 예보 배치. poll 하나가 곧 한 판이다.
 * 관측 INSERT 가 commit 된 뒤 별도 transaction 으로 돈다 —
 * 수집의 4단계 차감 사다리에 모델 추론 시간이 섞이지 않게 하기 위해서다.
 */
public class ForecastJob {

    private final RouteTargetQuery targets;                    // 예보가 안 붙은 poll 과 그 관측
    private final DemandProfileRepository profiles;
    private final ModelDeploymentRepository deployments;
    private final SeatForecastModel model;
    private final VehicleStopPredictionRepository predictions;
    private final Clock clock;

    // 생성자 생략

    @Transactional
    public void forecastPoll(long pollId) {

        // outcome 이 SUCCESS_ROWS·SUCCESS_EMPTY 인 판만 온다. SUCCESS_EMPTY 도 정상 스냅샷이다.
        PollSnapshot poll = targets.awaitingForecast(pollId);

        ModelDeployment deployment = deployments.active();   // 한 poll = 한 배포. transaction 내내 이 값을 쓴다
        Instant generatedAt = clock.instant();               // 한 poll = 같은 generatedAt

        // 셀 통계는 여기서 한 번만 읽는다. z화 · 이웃 폴백 · 구간합이 전부 이 객체 안에서 닫힌다.
        DemandProfile profile = profiles.load(
                poll.routeReferenceVersionId(),
                TimeBand.of(poll.responseReceivedAt()),
                deployment.featureContractVersion());

        List<VehicleStopPrediction> rows = new ArrayList<>();

        for (ObservedVehicle vehicle : poll.vehicles()) {
            if (vehicle.remainingSeats() == null) {
                continue;   // 좌석 결측 — 행을 만들 수 없다. 그 차량은 이 판의 arrivals 에서 사라진다(서빙 처리는 미결)
            }
            for (Horizon horizon : Horizon.all()) {
                int targetStopOrder = vehicle.stopOrder() + horizon.stops();
                if (targetStopOrder > poll.lastStopOrder()) {
                    break;      // 여정 끝을 넘어가는 대상은 만들지 않는다 — 종점 다음의 순번 1 은 다른 여정이다
                }
                if (!poll.boardingAllowed(targetStopOrder)) {
                    continue;   // 미정차 정류장은 행을 만들지 않는다. 사유는 route_stop 에 있으니 복제하지 않는다
                }
                SeatForecastResult result = model.predict(
                        new VehicleStopTarget(vehicle, targetStopOrder, horizon), profile);

                rows.add(VehicleStopPrediction.of(vehicle, targetStopOrder, horizon, result,
                        deployment.id(), profile.revision(), generatedAt));
            }
        }

        predictions.saveAll(rows);                                // 한 transaction
        predictions.markForecastCompleted(pollId, generatedAt);   // rows 가 비어도 찍는다
    }
}