/*
  FT546 shadow-only read model for SQL Server 2016 RTM (13.0).
  This package writes only alg.* and never changes Legacy WMS business rows.
  A clean install creates an INITIALIZING state for warehouse 01. The v3
  worker owns the first refresh and publication.
*/
SET NOCOUNT ON;
SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;

IF SCHEMA_ID(N'alg') IS NOT NULL
    THROW 51000, 'FT546 shadow install refused: alg schema already exists', 1;

CREATE SCHEMA alg;
GO

CREATE TABLE alg.shadow_package_manifest (
    package_id varchar(80) NOT NULL CONSTRAINT PK_alg_shadow_package_manifest PRIMARY KEY,
    package_version varchar(40) NOT NULL,
    installed_at datetime2 NOT NULL CONSTRAINT DF_alg_shadow_package_manifest_installed_at DEFAULT SYSUTCDATETIME()
);

INSERT INTO alg.shadow_package_manifest (package_id, package_version)
VALUES ('FT546WMS_SHADOW_ONLY', '2026-09-03-3');
GO

CREATE TABLE alg.warehouse_policy (
    warehouse_id varchar(80) NOT NULL PRIMARY KEY,
    target_fill_rate decimal(9,4) NULL,
    fifo_tolerance_hours int NULL,
    updated_at datetime2 NOT NULL CONSTRAINT DF_alg_warehouse_policy_updated_at DEFAULT SYSUTCDATETIME()
);
GO

CREATE TABLE alg.project_policy (
    project_no varchar(255) NOT NULL PRIMARY KEY,
    is_isolated bit NOT NULL CONSTRAINT DF_alg_project_policy_isolated DEFAULT 0,
    is_common bit NOT NULL CONSTRAINT DF_alg_project_policy_common DEFAULT 0,
    allow_mixed bit NOT NULL CONSTRAINT DF_alg_project_policy_mixed DEFAULT 0,
    updated_at datetime2 NOT NULL CONSTRAINT DF_alg_project_policy_updated_at DEFAULT SYSUTCDATETIME()
);
GO

CREATE TABLE alg.space_role (
    space_id varchar(80) NOT NULL PRIMARY KEY,
    role varchar(40) NOT NULL,
    updated_at datetime2 NOT NULL CONSTRAINT DF_alg_space_role_updated_at DEFAULT SYSUTCDATETIME(),
    CONSTRAINT CK_alg_space_role_role CHECK (role IN ('STORAGE', 'RELOCATION_BUFFER', 'OUTBOUND_BUFFER', 'PROTECTED_CORE'))
);
GO

CREATE TABLE alg.space_topology (
    warehouse_id varchar(80) NOT NULL,
    space_id varchar(80) NOT NULL,
    lane_code varchar(80) NOT NULL,
    lane_depth int NOT NULL,
    base_zone varchar(20) NOT NULL,
    snapshot_generation bigint NOT NULL,
    rebuilt_at datetime2 NOT NULL CONSTRAINT DF_alg_space_topology_rebuilt_at DEFAULT SYSUTCDATETIME(),
    CONSTRAINT PK_alg_space_topology PRIMARY KEY (warehouse_id, space_id),
    CONSTRAINT CK_alg_space_topology_base_zone CHECK (base_zone IN ('CACHE', 'RESERVE', 'MAIN', 'FLEX'))
);
GO

CREATE INDEX IX_alg_space_topology_warehouse_lane ON alg.space_topology(warehouse_id, lane_code);
GO

CREATE TABLE alg.inventory_item_snapshot (
    warehouse_id varchar(80) NOT NULL,
    stock_id varchar(80) NOT NULL,
    space_id varchar(80) NOT NULL,
    tray_code varchar(255) NULL,
    project_no varchar(255) NULL,
    product_no varchar(255) NULL,
    lot_no varchar(80) NULL,
    inbound_time_raw varchar(80) NULL,
    inbound_at datetime2 NULL,
    quantity decimal(38,6) NULL,
    out_quantity decimal(38,6) NULL,
    snapshot_generation bigint NOT NULL,
    synchronized_at datetime2 NOT NULL CONSTRAINT DF_alg_inventory_item_snapshot_sync DEFAULT SYSUTCDATETIME(),
    CONSTRAINT PK_alg_inventory_item_snapshot PRIMARY KEY (warehouse_id, stock_id)
);
GO

CREATE INDEX IX_alg_inventory_item_snapshot_space
    ON alg.inventory_item_snapshot(warehouse_id, space_id);
GO

CREATE INDEX IX_alg_inventory_item_snapshot_tray
    ON alg.inventory_item_snapshot(warehouse_id, tray_code)
    INCLUDE (space_id, project_no, product_no, snapshot_generation);
GO

CREATE INDEX IX_alg_inventory_item_snapshot_project_product
    ON alg.inventory_item_snapshot(warehouse_id, project_no, product_no)
    INCLUDE (space_id, tray_code, inbound_at, snapshot_generation);
GO

CREATE TABLE alg.space_inventory_summary (
    warehouse_id varchar(80) NOT NULL,
    space_id varchar(80) NOT NULL,
    stock_detail_count int NOT NULL,
    stock_tray_count int NOT NULL,
    stock_project_count int NOT NULL,
    stock_product_count int NOT NULL,
    snapshot_generation bigint NOT NULL,
    synchronized_at datetime2 NOT NULL CONSTRAINT DF_alg_space_inventory_summary_sync DEFAULT SYSUTCDATETIME(),
    CONSTRAINT PK_alg_space_inventory_summary PRIMARY KEY (warehouse_id, space_id)
);
GO

CREATE INDEX IX_alg_space_inventory_summary_space
    ON alg.space_inventory_summary(space_id, warehouse_id);
GO

CREATE TABLE alg.lane_state (
    warehouse_id varchar(80) NOT NULL,
    lane_code varchar(80) NOT NULL,
    space_count int NOT NULL CONSTRAINT DF_alg_lane_state_space_count DEFAULT 0,
    occupied_count int NOT NULL CONSTRAINT DF_alg_lane_state_occupied_count DEFAULT 0,
    available_count int NOT NULL CONSTRAINT DF_alg_lane_state_available_count DEFAULT 0,
    pending_outbound bit NOT NULL CONSTRAINT DF_alg_lane_state_pending_outbound DEFAULT 0,
    has_gap bit NOT NULL CONSTRAINT DF_alg_lane_state_has_gap DEFAULT 0,
    semantic_fingerprint varbinary(32) NULL,
    snapshot_generation bigint NOT NULL,
    state_version rowversion NOT NULL,
    rebuilt_at datetime2 NOT NULL CONSTRAINT DF_alg_lane_state_rebuilt_at DEFAULT SYSUTCDATETIME(),
    CONSTRAINT PK_alg_lane_state PRIMARY KEY (warehouse_id, lane_code)
);
GO

CREATE TABLE alg.reservation (
    reservation_id uniqueidentifier NOT NULL CONSTRAINT PK_alg_reservation PRIMARY KEY,
    warehouse_id varchar(80) NOT NULL,
    lane_code varchar(80) NOT NULL,
    space_id varchar(80) NOT NULL,
    project_no varchar(255) NULL,
    status varchar(20) NOT NULL,
    created_at datetime2 NOT NULL CONSTRAINT DF_alg_reservation_created_at DEFAULT SYSUTCDATETIME(),
    released_at datetime2 NULL,
    CONSTRAINT CK_alg_reservation_status CHECK (status IN ('ACTIVE', 'RELEASED', 'EXPIRED'))
);
GO

CREATE UNIQUE INDEX UX_alg_reservation_active_space
    ON alg.reservation(space_id) WHERE status = 'ACTIVE';
GO

CREATE TABLE alg.deep_workflow_lock (
    lock_id uniqueidentifier NOT NULL CONSTRAINT PK_alg_deep_workflow_lock PRIMARY KEY,
    request_id varchar(80) NOT NULL,
    warehouse_id varchar(80) NOT NULL,
    floor_no int NOT NULL,
    status varchar(20) NOT NULL,
    created_at datetime2 NOT NULL CONSTRAINT DF_alg_deep_workflow_lock_created_at DEFAULT SYSUTCDATETIME(),
    released_at datetime2 NULL,
    CONSTRAINT CK_alg_deep_workflow_lock_status CHECK (status IN ('ACTIVE', 'RELEASED'))
);
GO

CREATE UNIQUE INDEX UX_alg_deep_workflow_lock_active_floor
    ON alg.deep_workflow_lock(warehouse_id, floor_no) WHERE status = 'ACTIVE';
GO

CREATE TABLE alg.decision_log (
    decision_id uniqueidentifier NOT NULL CONSTRAINT PK_alg_decision_log PRIMARY KEY,
    warehouse_id varchar(80) NULL,
    decision_type varchar(40) NOT NULL,
    input_version binary(8) NULL,
    result_code varchar(80) NULL,
    reason nvarchar(4000) NULL,
    created_at datetime2 NOT NULL CONSTRAINT DF_alg_decision_log_created_at DEFAULT SYSUTCDATETIME()
);
GO

CREATE TABLE alg.shadow_snapshot_state (
    warehouse_id varchar(80) NOT NULL PRIMARY KEY,
    snapshot_generation bigint NOT NULL,
    snapshot_status varchar(20) NOT NULL,
    synchronized_at datetime2 NULL,
    source_signature varbinary(32) NULL,
    source_row_count bigint NULL,
    instance_id varchar(80) NULL,
    last_error varchar(1000) NULL,
    CONSTRAINT CK_alg_shadow_snapshot_state_status CHECK (snapshot_status IN ('INITIALIZING', 'REFRESHING', 'PUBLISHED', 'FAILED'))
);
GO

INSERT INTO alg.shadow_snapshot_state (warehouse_id, snapshot_generation, snapshot_status)
VALUES ('01', 0, 'INITIALIZING');
GO

CREATE TABLE alg.cargo_space_snapshot (
    warehouse_id varchar(80) NOT NULL,
    space_id varchar(80) NOT NULL,
    lane_code varchar(80) NULL,
    cargo_area_id varchar(80) NULL,
    roadway_id varchar(80) NULL,
    display_name varchar(255) NULL,
    platoon int NULL,
    column_no int NULL,
    floor_no int NULL,
    depth_raw varchar(80) NULL,
    space_status varchar(20) NULL,
    locked varchar(20) NULL,
    tray_code varchar(255) NULL,
    snapshot_generation bigint NOT NULL,
    synchronized_at datetime2 NOT NULL,
    CONSTRAINT PK_alg_cargo_space_snapshot PRIMARY KEY (warehouse_id, space_id)
);
GO

CREATE TABLE alg.active_job_snapshot (
    warehouse_id varchar(80) NOT NULL,
    job_id varchar(80) NOT NULL,
    task_no varchar(80) NULL,
    tray_code varchar(255) NULL,
    source_space_id varchar(80) NULL,
    target_space_id varchar(80) NULL,
    status varchar(20) NULL,
    in_or_out varchar(20) NULL,
    create_time_raw varchar(80) NULL,
    snapshot_generation bigint NOT NULL,
    synchronized_at datetime2 NOT NULL,
    CONSTRAINT PK_alg_active_job_snapshot PRIMARY KEY (warehouse_id, job_id)
);
GO

CREATE TABLE alg.shadow_inbound_request_snapshot (
    warehouse_id varchar(80) NOT NULL,
    task_no varchar(80) NOT NULL,
    tray_no varchar(255) NOT NULL,
    request_id varchar(80) NULL,
    raw_warehouse_id varchar(80) NULL,
    project_no varchar(255) NULL,
    product_id varchar(255) NULL,
    lot_no varchar(80) NULL,
    snapshot_generation bigint NOT NULL,
    synchronized_at datetime2 NOT NULL,
    CONSTRAINT PK_alg_shadow_inbound_request_snapshot PRIMARY KEY (warehouse_id, task_no, tray_no)
);
GO

CREATE TABLE alg.shadow_outbound_request_snapshot (
    warehouse_id varchar(80) NOT NULL,
    stockout_detail_id varchar(80) NOT NULL,
    raw_warehouse_id varchar(80) NULL,
    project_no varchar(255) NULL,
    product_id varchar(255) NULL,
    requested_quantity decimal(38,6) NULL,
    snapshot_generation bigint NOT NULL,
    synchronized_at datetime2 NOT NULL,
    CONSTRAINT PK_alg_shadow_outbound_request_snapshot PRIMARY KEY (warehouse_id, stockout_detail_id)
);
GO

CREATE TABLE alg.shadow_runtime_metric (
    warehouse_id varchar(80) NOT NULL,
    metric_name varchar(80) NOT NULL,
    bucket_upper_micros bigint NOT NULL,
    metric_count bigint NOT NULL CONSTRAINT DF_alg_shadow_runtime_metric_count DEFAULT 0,
    metric_total_micros bigint NOT NULL CONSTRAINT DF_alg_shadow_runtime_metric_total DEFAULT 0,
    metric_max_micros bigint NOT NULL CONSTRAINT DF_alg_shadow_runtime_metric_max DEFAULT 0,
    flushed_at datetime2 NULL,
    CONSTRAINT PK_alg_shadow_runtime_metric PRIMARY KEY (warehouse_id, metric_name, bucket_upper_micros)
);
GO

CREATE TABLE alg.shadow_instance_heartbeat (
    warehouse_id varchar(80) NOT NULL PRIMARY KEY,
    instance_id varchar(80) NOT NULL,
    claimed_at datetime2 NOT NULL,
    last_seen_at datetime2 NOT NULL
);
GO

CREATE PROCEDURE alg.usp_rebuild_space_topology
    @warehouse_id varchar(80) = NULL,
    @snapshot_generation bigint = NULL
AS
BEGIN
    SET NOCOUNT ON;
    IF NULLIF(LTRIM(RTRIM(@warehouse_id)), '') IS NULL
        THROW 51019, 'warehouse_id is required', 1;
    DECLARE @generation bigint;

    SET @generation = @snapshot_generation;
    IF @generation IS NULL
        SELECT @generation = state.snapshot_generation
        FROM alg.shadow_snapshot_state state
        WHERE state.warehouse_id = @warehouse_id
          AND state.snapshot_status = 'PUBLISHED';

    IF @generation IS NULL
        THROW 51020, 'FT546 shadow topology rebuild requires a snapshot generation', 1;

    ;WITH lane_topology AS (
        SELECT cargo.warehouse_id, cargo.lane_code,
               COUNT(*) AS lane_depth,
               MAX(cargo.column_no) AS column_no,
               MIN(cargo.platoon) AS row_no,
               MIN(cargo.floor_no) AS floor_no,
               MIN(cargo.space_id) AS stable_id
        FROM alg.cargo_space_snapshot cargo
        WHERE cargo.snapshot_generation = @generation
          AND NULLIF(LTRIM(RTRIM(cargo.lane_code)), '') IS NOT NULL
          AND TRY_CONVERT(int, cargo.depth_raw) IS NOT NULL
          AND (@warehouse_id IS NULL OR cargo.warehouse_id = @warehouse_id)
        GROUP BY cargo.warehouse_id, cargo.lane_code
    ), ranked_lanes AS (
        SELECT lane_topology.*,
               CASE WHEN lane_depth = 1 THEN ROW_NUMBER() OVER (
                   PARTITION BY warehouse_id
                   ORDER BY column_no DESC, row_no ASC, floor_no ASC, stable_id ASC
               ) END AS single_depth_rank
        FROM lane_topology
    ), source_rows AS (
        SELECT cargo.warehouse_id, cargo.space_id, cargo.lane_code,
               ranked_lanes.lane_depth,
               CASE WHEN ranked_lanes.lane_depth = 1 AND ranked_lanes.single_depth_rank <= 80 THEN 'CACHE'
                    WHEN ranked_lanes.lane_depth = 1 AND ranked_lanes.single_depth_rank <= 133 THEN 'RESERVE'
                    WHEN ranked_lanes.lane_depth = 5 THEN 'MAIN'
                    ELSE 'FLEX' END AS base_zone
        FROM alg.cargo_space_snapshot cargo
        JOIN ranked_lanes ON ranked_lanes.warehouse_id = cargo.warehouse_id
            AND ranked_lanes.lane_code = cargo.lane_code
        WHERE cargo.snapshot_generation = @generation
          AND NULLIF(LTRIM(RTRIM(cargo.lane_code)), '') IS NOT NULL
          AND TRY_CONVERT(int, cargo.depth_raw) IS NOT NULL
          AND (@warehouse_id IS NULL OR cargo.warehouse_id = @warehouse_id)
    )
    MERGE alg.space_topology WITH (HOLDLOCK) AS target
    USING source_rows AS source
      ON target.warehouse_id = source.warehouse_id AND target.space_id = source.space_id
    WHEN MATCHED AND (target.lane_code <> source.lane_code OR target.lane_depth <> source.lane_depth
        OR target.base_zone <> source.base_zone OR target.snapshot_generation <> @generation)
      THEN UPDATE SET lane_code = source.lane_code, lane_depth = source.lane_depth,
                      base_zone = source.base_zone, snapshot_generation = @generation,
                      rebuilt_at = SYSUTCDATETIME()
    WHEN NOT MATCHED BY TARGET THEN
      INSERT (warehouse_id, space_id, lane_code, lane_depth, base_zone, snapshot_generation)
      VALUES (source.warehouse_id, source.space_id, source.lane_code, source.lane_depth,
              source.base_zone, @generation)
    WHEN NOT MATCHED BY SOURCE AND (@warehouse_id IS NULL OR target.warehouse_id = @warehouse_id)
      THEN DELETE;
END;
GO

CREATE PROCEDURE alg.usp_sync_inventory_snapshot
    @warehouse_id varchar(80) = NULL,
    @snapshot_generation bigint = NULL
AS
BEGIN
    SET NOCOUNT ON;
    IF NULLIF(LTRIM(RTRIM(@warehouse_id)), '') IS NULL
        THROW 51019, 'warehouse_id is required', 1;
    DECLARE @generation bigint;

    SET @generation = @snapshot_generation;
    IF @generation IS NULL
        SELECT @generation = state.snapshot_generation
        FROM alg.shadow_snapshot_state state
        WHERE state.warehouse_id = @warehouse_id
          AND state.snapshot_status = 'PUBLISHED';

    IF @generation IS NULL
        THROW 51021, 'FT546 shadow inventory summary requires a snapshot generation', 1;

    DELETE summary
    FROM alg.space_inventory_summary summary
    WHERE (@warehouse_id IS NULL OR summary.warehouse_id = @warehouse_id);

    INSERT INTO alg.space_inventory_summary (
        warehouse_id, space_id, stock_detail_count, stock_tray_count,
        stock_project_count, stock_product_count, snapshot_generation, synchronized_at)
    SELECT item.warehouse_id, item.space_id, COUNT(*),
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(item.tray_code)), '')),
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(item.project_no)), '')),
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(item.product_no)), '')),
           @generation, SYSUTCDATETIME()
    FROM alg.inventory_item_snapshot item
    WHERE item.snapshot_generation = @generation
      AND (@warehouse_id IS NULL OR item.warehouse_id = @warehouse_id)
    GROUP BY item.warehouse_id, item.space_id;
END;
GO

CREATE VIEW alg.v_inventory_item
AS
-- generation contract: WHERE snapshot_generation = published_snapshot.snapshot_generation
SELECT item.warehouse_id, item.stock_id, item.space_id, item.tray_code,
       item.project_no, item.product_no, item.lot_no, item.inbound_time_raw,
       item.inbound_at, item.quantity, item.out_quantity, item.snapshot_generation
FROM alg.inventory_item_snapshot item
JOIN alg.shadow_snapshot_state published_snapshot
  ON published_snapshot.warehouse_id = item.warehouse_id
 AND published_snapshot.snapshot_status = 'PUBLISHED'
 AND item.snapshot_generation = published_snapshot.snapshot_generation;
GO

CREATE VIEW alg.v_active_job
AS
WITH published_jobs AS (
    SELECT job.warehouse_id, job.job_id, job.task_no, job.tray_code,
           job.source_space_id, job.target_space_id, job.status, job.in_or_out,
           job.create_time_raw, job.snapshot_generation
    FROM alg.active_job_snapshot job
    JOIN alg.shadow_snapshot_state published_snapshot
      ON published_snapshot.warehouse_id = job.warehouse_id
     AND published_snapshot.snapshot_status = 'PUBLISHED'
     AND job.snapshot_generation = published_snapshot.snapshot_generation
), tray_candidates AS (
    SELECT job.warehouse_id, job.job_id, job.snapshot_generation,
           COUNT(DISTINCT stock.space_id) AS space_count,
           MIN(stock.space_id) AS cargo_space_id,
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(stock.project_no)), '')) AS project_count,
           MIN(NULLIF(LTRIM(RTRIM(stock.project_no)), '')) AS project_no,
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(stock.product_no)), '')) AS product_count,
           MIN(NULLIF(LTRIM(RTRIM(stock.product_no)), '')) AS product_no
    FROM published_jobs job
    JOIN alg.inventory_item_snapshot stock
      ON stock.warehouse_id = job.warehouse_id
     AND stock.snapshot_generation = job.snapshot_generation
     AND stock.tray_code = NULLIF(LTRIM(RTRIM(job.tray_code)), '')
    GROUP BY job.warehouse_id, job.job_id, job.snapshot_generation
), resolved_jobs AS (
    SELECT job.job_id, job.task_no, job.tray_code, job.target_space_id,
           job.warehouse_id, job.status, job.in_or_out, job.create_time_raw,
           job.snapshot_generation,
           COALESCE(NULLIF(LTRIM(RTRIM(job.source_space_id)), ''),
                    CASE WHEN tray_candidates.space_count = 1 THEN tray_candidates.cargo_space_id END) AS source_space_id,
           CASE WHEN NULLIF(LTRIM(RTRIM(job.source_space_id)), '') IS NOT NULL THEN 'EXPLICIT'
                WHEN tray_candidates.space_count = 1 THEN 'TRAY'
                WHEN tray_candidates.space_count > 1 THEN 'AMBIGUOUS_TRAY'
                ELSE 'UNRESOLVED' END AS source_resolution
    FROM published_jobs job
    LEFT JOIN tray_candidates
      ON tray_candidates.warehouse_id = job.warehouse_id
     AND tray_candidates.job_id = job.job_id
     AND tray_candidates.snapshot_generation = job.snapshot_generation
    WHERE job.snapshot_generation IS NOT NULL
), source_scope AS (
    SELECT resolved.warehouse_id, resolved.job_id, resolved.snapshot_generation,
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(stock.project_no)), '')) AS project_count,
           MIN(NULLIF(LTRIM(RTRIM(stock.project_no)), '')) AS project_no,
           COUNT(DISTINCT NULLIF(LTRIM(RTRIM(stock.product_no)), '')) AS product_count,
           MIN(NULLIF(LTRIM(RTRIM(stock.product_no)), '')) AS product_no
    FROM resolved_jobs resolved
    JOIN alg.inventory_item_snapshot stock
      ON stock.warehouse_id = resolved.warehouse_id
     AND stock.snapshot_generation = resolved.snapshot_generation
     AND stock.space_id = resolved.source_space_id
    GROUP BY resolved.warehouse_id, resolved.job_id, resolved.snapshot_generation
)
SELECT resolved.job_id, resolved.task_no, resolved.tray_code,
       CASE WHEN resolved.source_space_id IS NOT NULL
            THEN CASE WHEN source_scope.project_count = 1 THEN source_scope.project_no END
            ELSE CASE WHEN tray_candidates.project_count = 1 THEN tray_candidates.project_no END END AS project_no,
       CASE WHEN resolved.source_space_id IS NOT NULL
            THEN CASE WHEN source_scope.product_count = 1 THEN source_scope.product_no END
            ELSE CASE WHEN tray_candidates.product_count = 1 THEN tray_candidates.product_no END END AS product_no,
       resolved.source_space_id, resolved.target_space_id AS target_cargo_space_id,
       resolved.warehouse_id, resolved.status, resolved.in_or_out,
       resolved.create_time_raw, source_cargo.lane_code AS source_lane_code,
       resolved.source_resolution, resolved.snapshot_generation
FROM resolved_jobs resolved
 LEFT JOIN tray_candidates
  ON tray_candidates.warehouse_id = resolved.warehouse_id
 AND tray_candidates.job_id = resolved.job_id
 AND tray_candidates.snapshot_generation = resolved.snapshot_generation
 LEFT JOIN source_scope
  ON source_scope.warehouse_id = resolved.warehouse_id
 AND source_scope.job_id = resolved.job_id
 AND source_scope.snapshot_generation = resolved.snapshot_generation
LEFT JOIN alg.cargo_space_snapshot source_cargo
  ON source_cargo.warehouse_id = resolved.warehouse_id
 AND source_cargo.space_id = resolved.source_space_id
 AND source_cargo.snapshot_generation = resolved.snapshot_generation;
GO

CREATE VIEW alg.v_space_state
AS
WITH published_snapshot AS (
    SELECT state.warehouse_id, state.snapshot_generation
    FROM alg.shadow_snapshot_state state
    WHERE state.snapshot_status = 'PUBLISHED'
), outbound_lanes AS (
    SELECT DISTINCT job.warehouse_id, job.source_lane_code AS lane_code,
           job.snapshot_generation
    FROM alg.v_active_job job
    WHERE TRY_CONVERT(int, job.in_or_out) >= 20
      AND job.source_lane_code IS NOT NULL
), lane_gaps AS (
    SELECT DISTINCT inner_space.warehouse_id, inner_space.lane_code,
           inner_space.snapshot_generation
    FROM alg.cargo_space_snapshot inner_space
    JOIN alg.cargo_space_snapshot outer_space
      ON outer_space.warehouse_id = inner_space.warehouse_id
     AND outer_space.lane_code = inner_space.lane_code
     AND outer_space.snapshot_generation = inner_space.snapshot_generation
     AND TRY_CONVERT(int, outer_space.depth_raw) > TRY_CONVERT(int, inner_space.depth_raw)
    JOIN published_snapshot published
      ON published.warehouse_id = inner_space.warehouse_id
     AND published.snapshot_generation = inner_space.snapshot_generation
    LEFT JOIN alg.space_inventory_summary inner_stock
      ON inner_stock.warehouse_id = inner_space.warehouse_id
     AND inner_stock.space_id = inner_space.space_id
     AND inner_stock.snapshot_generation = inner_space.snapshot_generation
    LEFT JOIN alg.space_inventory_summary outer_stock
      ON outer_stock.warehouse_id = outer_space.warehouse_id
     AND outer_stock.space_id = outer_space.space_id
     AND outer_stock.snapshot_generation = outer_space.snapshot_generation
    WHERE TRY_CONVERT(int, inner_space.depth_raw) IS NOT NULL
      AND TRY_CONVERT(int, outer_space.depth_raw) IS NOT NULL
      AND NULLIF(LTRIM(RTRIM(inner_space.tray_code)), '') IS NULL
      AND inner_stock.space_id IS NULL
      AND (NULLIF(LTRIM(RTRIM(outer_space.tray_code)), '') IS NOT NULL
           OR outer_stock.space_id IS NOT NULL)
)
SELECT cargo.space_id, cargo.warehouse_id, cargo.lane_code,
       cargo.cargo_area_id AS area_id,
       CASE WHEN ISNULL(role.role, 'STORAGE') IN ('RELOCATION_BUFFER', 'OUTBOUND_BUFFER') THEN 'CACHE'
            WHEN ISNULL(role.role, 'STORAGE') = 'PROTECTED_CORE' THEN 'RESERVE'
            ELSE topology.base_zone END AS zone_code,
       cargo.roadway_id, cargo.display_name, cargo.platoon, cargo.column_no,
       cargo.floor_no, cargo.depth_raw,
       TRY_CONVERT(int, cargo.depth_raw) AS depth, cargo.space_status,
       cargo.locked, cargo.tray_code AS space_tray_code,
       ISNULL(role.role, 'STORAGE') AS role,
       ISNULL(stock.stock_detail_count, 0) AS stock_detail_count,
       ISNULL(stock.stock_tray_count, 0) AS stock_tray_count,
       ISNULL(stock.stock_project_count, 0) AS stock_project_count,
       ISNULL(stock.stock_product_count, 0) AS stock_product_count,
       CONVERT(bit, CASE WHEN ISNULL(stock.stock_tray_count, 0) > 1 THEN 1 ELSE 0 END) AS tray_conflict,
       CONVERT(bit, CASE WHEN ISNULL(stock.stock_project_count, 0) > 1 THEN 1 ELSE 0 END) AS project_conflict,
       CONVERT(bit, 0) AS product_conflict,
       CONVERT(bit, CASE WHEN outbound_lanes.lane_code IS NULL THEN 0 ELSE 1 END) AS pending_outbound,
       active_reservations.reservation_id,
       CONVERT(bit, CASE WHEN active_reservations.reservation_id IS NULL THEN 0 ELSE 1 END) AS reserved,
       CONVERT(bit, CASE WHEN lane_gaps.lane_code IS NULL THEN 0 ELSE 1 END) AS has_gap,
       CONVERT(bit, CASE
           WHEN cargo.space_status IS NULL OR cargo.space_status <> '1' THEN 0
           WHEN NULLIF(LTRIM(RTRIM(cargo.locked)), '') IS NOT NULL
                AND LTRIM(RTRIM(cargo.locked)) NOT IN ('0', 'N') THEN 0
           WHEN NULLIF(LTRIM(RTRIM(cargo.tray_code)), '') IS NOT NULL THEN 0
           WHEN ISNULL(stock.stock_detail_count, 0) <> 0 THEN 0
           WHEN TRY_CONVERT(int, cargo.depth_raw) IS NULL THEN 0
           WHEN ISNULL(role.role, 'STORAGE') <> 'STORAGE' THEN 0
           WHEN outbound_lanes.lane_code IS NOT NULL THEN 0
           WHEN active_reservations.reservation_id IS NOT NULL THEN 0
           WHEN EXISTS (
               SELECT 1
               FROM alg.cargo_space_snapshot inner_candidate
               LEFT JOIN alg.space_inventory_summary inner_candidate_stock
                 ON inner_candidate_stock.warehouse_id = inner_candidate.warehouse_id
                AND inner_candidate_stock.space_id = inner_candidate.space_id
                AND inner_candidate_stock.snapshot_generation = inner_candidate.snapshot_generation
               WHERE inner_candidate.warehouse_id = cargo.warehouse_id
                 AND inner_candidate.lane_code = cargo.lane_code
                 AND inner_candidate.snapshot_generation = cargo.snapshot_generation
                 AND TRY_CONVERT(int, inner_candidate.depth_raw) < TRY_CONVERT(int, cargo.depth_raw)
                 AND inner_candidate.space_status = '1'
                 AND NULLIF(LTRIM(RTRIM(inner_candidate.tray_code)), '') IS NULL
                 AND inner_candidate_stock.space_id IS NULL
           ) THEN 0
           ELSE 1 END) AS inbound_available,
       cargo.snapshot_generation
FROM alg.cargo_space_snapshot cargo
JOIN published_snapshot published
  ON published.warehouse_id = cargo.warehouse_id
 AND published.snapshot_generation = cargo.snapshot_generation
JOIN alg.space_topology topology
  ON topology.warehouse_id = cargo.warehouse_id
 AND topology.space_id = cargo.space_id
 AND topology.snapshot_generation = cargo.snapshot_generation
LEFT JOIN alg.space_inventory_summary stock
  ON stock.warehouse_id = cargo.warehouse_id
 AND stock.space_id = cargo.space_id
 AND stock.snapshot_generation = cargo.snapshot_generation
LEFT JOIN alg.space_role role ON role.space_id = cargo.space_id
LEFT JOIN outbound_lanes
  ON outbound_lanes.warehouse_id = cargo.warehouse_id
 AND outbound_lanes.lane_code = cargo.lane_code
 AND outbound_lanes.snapshot_generation = cargo.snapshot_generation
LEFT JOIN alg.reservation active_reservations
  ON active_reservations.space_id = cargo.space_id
 AND active_reservations.status = 'ACTIVE'
LEFT JOIN lane_gaps
  ON lane_gaps.warehouse_id = cargo.warehouse_id
 AND lane_gaps.lane_code = cargo.lane_code
 AND lane_gaps.snapshot_generation = cargo.snapshot_generation;
GO

CREATE PROCEDURE alg.usp_compute_lane_semantic_fingerprint
    @warehouse_id varchar(80),
    @lane_code varchar(80),
    @semantic_fingerprint varbinary(32) OUTPUT,
    @inventory_source varchar(20) = 'SNAPSHOT',
    @snapshot_generation bigint = NULL
AS
BEGIN
    SET NOCOUNT ON;
    DECLARE @generation bigint, @payload nvarchar(max);

    IF @inventory_source <> 'SNAPSHOT'
        THROW 51022, 'FT546 shadow fingerprint source must be SNAPSHOT', 1;

    SELECT @generation = COALESCE(@snapshot_generation, state.snapshot_generation)
    FROM alg.shadow_snapshot_state state
    WHERE state.warehouse_id = @warehouse_id
      AND state.snapshot_status = 'PUBLISHED';

    IF @generation IS NULL
        THROW 51023, 'FT546 shadow fingerprint requires a snapshot generation', 1;

    SELECT @payload = ISNULL(CONVERT(nvarchar(max), (
        SELECT cargo.space_id AS [@space_id], cargo.space_status AS [@status],
               cargo.locked AS [@locked], cargo.tray_code AS [@tray_code],
               cargo.depth_raw AS [@depth], cargo.platoon AS [@platoon],
               cargo.column_no AS [@column], cargo.floor_no AS [@floor],
               topology.lane_depth AS [@lane_depth], topology.base_zone AS [@base_zone],
               ISNULL(role.role, 'STORAGE') AS [@role]
        FROM alg.cargo_space_snapshot cargo
        LEFT JOIN alg.space_topology topology
          ON topology.warehouse_id = cargo.warehouse_id
         AND topology.space_id = cargo.space_id
         AND topology.snapshot_generation = @generation
        LEFT JOIN alg.space_role role ON role.space_id = cargo.space_id
        WHERE cargo.warehouse_id = @warehouse_id
          AND cargo.lane_code = @lane_code
          AND cargo.snapshot_generation = @generation
        ORDER BY TRY_CONVERT(int, cargo.depth_raw), cargo.space_id
        FOR XML PATH('space'), ROOT('spaces'), TYPE
    )), '');

    SET @payload = @payload + '|INVENTORY|' + ISNULL(CONVERT(nvarchar(max), (
        SELECT item.stock_id AS [@stock_id], item.space_id AS [@space_id],
               item.tray_code AS [@tray_code], item.project_no AS [@project_no],
               item.product_no AS [@product_no], item.lot_no AS [@lot_no],
               item.inbound_time_raw AS [@inbound_time], item.quantity AS [@quantity],
               item.out_quantity AS [@out_quantity]
        FROM alg.inventory_item_snapshot item
        WHERE item.warehouse_id = @warehouse_id
          AND item.snapshot_generation = @generation
          AND EXISTS (SELECT 1 FROM alg.cargo_space_snapshot cargo
                      WHERE cargo.warehouse_id = item.warehouse_id
                        AND cargo.space_id = item.space_id
                        AND cargo.lane_code = @lane_code
                        AND cargo.snapshot_generation = @generation)
        ORDER BY item.space_id, item.stock_id
        FOR XML PATH('item'), ROOT('inventory'), TYPE
    )), '');

    SET @payload = @payload + '|ACTIVE_JOBS|' + ISNULL(CONVERT(nvarchar(max), (
        SELECT job.job_id AS [@job_id], job.task_no AS [@task_no], job.tray_code AS [@tray_code],
               job.source_space_id AS [@source_space_id], job.target_space_id AS [@target_space_id],
               job.status AS [@status], job.in_or_out AS [@in_or_out],
               job.create_time_raw AS [@create_time]
        FROM alg.active_job_snapshot job
        JOIN alg.cargo_space_snapshot source_cargo
          ON source_cargo.warehouse_id = job.warehouse_id
         AND source_cargo.space_id = job.source_space_id
         AND source_cargo.snapshot_generation = job.snapshot_generation
        WHERE job.warehouse_id = @warehouse_id
          AND source_cargo.lane_code = @lane_code
          AND job.snapshot_generation = @generation
        ORDER BY job.job_id
        FOR XML PATH('job'), ROOT('jobs'), TYPE
    )), '');

    SET @payload = @payload + '|PROJECT_POLICY|' + ISNULL(CONVERT(nvarchar(max), (
        SELECT policy.project_no AS [@project_no], policy.is_isolated AS [@is_isolated],
               policy.is_common AS [@is_common], policy.allow_mixed AS [@allow_mixed]
        FROM alg.project_policy policy
        ORDER BY policy.project_no
        FOR XML PATH('project'), ROOT('project_policy'), TYPE
    )), '');

    SET @payload = @payload + '|WAREHOUSE_POLICY|' + ISNULL(CONVERT(nvarchar(max), (
        SELECT policy.warehouse_id AS [@warehouse_id], policy.target_fill_rate AS [@target_fill_rate],
               policy.fifo_tolerance_hours AS [@fifo_tolerance_hours]
        FROM alg.warehouse_policy policy
        WHERE policy.warehouse_id = @warehouse_id
        FOR XML PATH('warehouse'), ROOT('warehouse_policy'), TYPE
    )), '');

    SET @payload = @payload + '|RESERVATIONS|' + ISNULL(CONVERT(nvarchar(max), (
        SELECT reservation.reservation_id AS [@reservation_id], reservation.space_id AS [@space_id],
               reservation.project_no AS [@project_no], reservation.status AS [@status]
        FROM alg.reservation reservation
        WHERE reservation.warehouse_id = @warehouse_id
          AND reservation.lane_code = @lane_code
          AND reservation.status = 'ACTIVE'
        ORDER BY reservation.space_id, reservation.reservation_id
        FOR XML PATH('reservation'), ROOT('reservations'), TYPE
    )), '');

    SET @semantic_fingerprint = HASHBYTES('SHA2_256', CONVERT(varbinary(max), @payload));
END;
GO

CREATE PROCEDURE alg.usp_rebuild_lane_state
    @warehouse_id varchar(80) = NULL,
    @lane_code varchar(80) = NULL,
    @snapshot_generation bigint = NULL
AS
BEGIN
    SET NOCOUNT ON;
    IF NULLIF(LTRIM(RTRIM(@warehouse_id)), '') IS NULL
        THROW 51019, 'warehouse_id is required', 1;
    DECLARE @generation bigint;

    SET @generation = @snapshot_generation;
    IF @generation IS NULL
        SELECT @generation = state.snapshot_generation
        FROM alg.shadow_snapshot_state state
        WHERE state.warehouse_id = @warehouse_id
          AND state.snapshot_status = 'PUBLISHED';

    IF @generation IS NULL
        THROW 51024, 'FT546 shadow lane rebuild requires a snapshot generation', 1;

     ;WITH outbound_lanes AS (
         SELECT DISTINCT job.warehouse_id, cargo.lane_code
         FROM alg.active_job_snapshot job
         JOIN alg.cargo_space_snapshot cargo
           ON cargo.warehouse_id = job.warehouse_id
          AND cargo.space_id = job.source_space_id
          AND cargo.snapshot_generation = @generation
         WHERE job.snapshot_generation = @generation
           AND TRY_CONVERT(int, job.in_or_out) >= 20
           AND cargo.lane_code IS NOT NULL
    ), lane_gaps AS (
        SELECT DISTINCT inner_space.warehouse_id, inner_space.lane_code
        FROM alg.cargo_space_snapshot inner_space
        JOIN alg.cargo_space_snapshot outer_space
          ON outer_space.warehouse_id = inner_space.warehouse_id
         AND outer_space.lane_code = inner_space.lane_code
         AND outer_space.snapshot_generation = @generation
         AND TRY_CONVERT(int, outer_space.depth_raw) > TRY_CONVERT(int, inner_space.depth_raw)
        LEFT JOIN alg.space_inventory_summary inner_stock
          ON inner_stock.warehouse_id = inner_space.warehouse_id
         AND inner_stock.space_id = inner_space.space_id
         AND inner_stock.snapshot_generation = @generation
        LEFT JOIN alg.space_inventory_summary outer_stock
          ON outer_stock.warehouse_id = outer_space.warehouse_id
         AND outer_stock.space_id = outer_space.space_id
         AND outer_stock.snapshot_generation = @generation
        WHERE inner_space.snapshot_generation = @generation
          AND NULLIF(LTRIM(RTRIM(inner_space.tray_code)), '') IS NULL
          AND inner_stock.space_id IS NULL
          AND (NULLIF(LTRIM(RTRIM(outer_space.tray_code)), '') IS NOT NULL
               OR outer_stock.space_id IS NOT NULL)
    )
    SELECT cargo.warehouse_id, cargo.lane_code,
            COUNT(*) AS space_count,
           SUM(CASE WHEN cargo.space_status = '2' OR ISNULL(stock.stock_detail_count, 0) > 0 THEN 1 ELSE 0 END) AS occupied_count,
           SUM(CASE WHEN cargo.space_status = '1'
                     AND (NULLIF(LTRIM(RTRIM(cargo.locked)), '') IS NULL OR LTRIM(RTRIM(cargo.locked)) IN ('0', 'N'))
                     AND NULLIF(LTRIM(RTRIM(cargo.tray_code)), '') IS NULL
                     AND ISNULL(stock.stock_detail_count, 0) = 0
                     AND ISNULL(role.role, 'STORAGE') = 'STORAGE'
                     AND outbound_lanes.lane_code IS NULL
                     AND reservation.reservation_id IS NULL
                     AND lane_gaps.lane_code IS NULL
                    THEN 1 ELSE 0 END) AS available_count,
           CONVERT(bit, CASE WHEN outbound_lanes.lane_code IS NULL THEN 0 ELSE 1 END) AS pending_outbound,
            CONVERT(bit, CASE WHEN lane_gaps.lane_code IS NULL THEN 0 ELSE 1 END) AS has_gap,
            CAST(NULL AS varbinary(32)) AS semantic_fingerprint
    INTO #source_state
    FROM alg.cargo_space_snapshot cargo
    LEFT JOIN alg.space_inventory_summary stock
      ON stock.warehouse_id = cargo.warehouse_id
     AND stock.space_id = cargo.space_id
     AND stock.snapshot_generation = @generation
    LEFT JOIN alg.space_role role ON role.space_id = cargo.space_id
    LEFT JOIN outbound_lanes
      ON outbound_lanes.warehouse_id = cargo.warehouse_id
     AND outbound_lanes.lane_code = cargo.lane_code
    LEFT JOIN alg.reservation reservation
      ON reservation.space_id = cargo.space_id
     AND reservation.status = 'ACTIVE'
    LEFT JOIN lane_gaps
      ON lane_gaps.warehouse_id = cargo.warehouse_id
     AND lane_gaps.lane_code = cargo.lane_code
    WHERE cargo.snapshot_generation = @generation
      AND (@warehouse_id IS NULL OR cargo.warehouse_id = @warehouse_id)
      AND (@lane_code IS NULL OR cargo.lane_code = @lane_code)
    GROUP BY cargo.warehouse_id, cargo.lane_code,
             outbound_lanes.lane_code, lane_gaps.lane_code;

    DECLARE @fingerprint_warehouse varchar(80), @fingerprint_lane varchar(80),
            @fingerprint varbinary(32);
    DECLARE fingerprint_cursor CURSOR LOCAL FAST_FORWARD FOR
        SELECT warehouse_id, lane_code FROM #source_state ORDER BY warehouse_id, lane_code;
    OPEN fingerprint_cursor;
    FETCH NEXT FROM fingerprint_cursor INTO @fingerprint_warehouse, @fingerprint_lane;
    WHILE @@FETCH_STATUS = 0
    BEGIN
        EXEC alg.usp_compute_lane_semantic_fingerprint
            @warehouse_id = @fingerprint_warehouse,
            @lane_code = @fingerprint_lane,
            @semantic_fingerprint = @fingerprint OUTPUT,
            @inventory_source = 'SNAPSHOT',
            @snapshot_generation = @generation;
        UPDATE #source_state SET semantic_fingerprint = @fingerprint
        WHERE warehouse_id = @fingerprint_warehouse AND lane_code = @fingerprint_lane;
        FETCH NEXT FROM fingerprint_cursor INTO @fingerprint_warehouse, @fingerprint_lane;
    END;
    CLOSE fingerprint_cursor;
    DEALLOCATE fingerprint_cursor;

    MERGE alg.lane_state WITH (HOLDLOCK) AS target
    USING #source_state AS source
      ON target.warehouse_id = source.warehouse_id AND target.lane_code = source.lane_code
    WHEN MATCHED AND (target.space_count <> source.space_count OR
                      target.occupied_count <> source.occupied_count OR
                      target.available_count <> source.available_count OR
                      target.pending_outbound <> source.pending_outbound OR
                      target.has_gap <> source.has_gap OR
                      target.snapshot_generation <> @generation OR
                      target.semantic_fingerprint IS NULL OR
                      target.semantic_fingerprint <> source.semantic_fingerprint)
      THEN UPDATE SET space_count = source.space_count,
                      occupied_count = source.occupied_count,
                      available_count = source.available_count,
                      pending_outbound = source.pending_outbound,
                      has_gap = source.has_gap,
                      semantic_fingerprint = source.semantic_fingerprint,
                      snapshot_generation = @generation,
                      rebuilt_at = SYSUTCDATETIME()
    WHEN NOT MATCHED BY TARGET THEN
      INSERT (warehouse_id, lane_code, space_count, occupied_count, available_count,
              pending_outbound, has_gap, semantic_fingerprint, snapshot_generation)
      VALUES (source.warehouse_id, source.lane_code, source.space_count, source.occupied_count,
              source.available_count, source.pending_outbound, source.has_gap,
              source.semantic_fingerprint, @generation)
    WHEN NOT MATCHED BY SOURCE
      AND (@warehouse_id IS NULL OR target.warehouse_id = @warehouse_id)
      AND (@lane_code IS NULL OR target.lane_code = @lane_code)
      THEN DELETE;
END;
GO

CREATE PROCEDURE alg.usp_refresh_dirty_lanes
    @warehouse_id varchar(80)
AS
BEGIN
    SET NOCOUNT ON;
    IF NULLIF(LTRIM(RTRIM(@warehouse_id)), '') IS NULL
        THROW 51002, 'warehouse_id is required', 1;

    DECLARE @generation bigint;
    SELECT @generation = snapshot_generation
    FROM alg.shadow_snapshot_state
    WHERE warehouse_id = @warehouse_id AND snapshot_status = 'PUBLISHED';

    IF @generation IS NULL
        THROW 51003, 'published snapshot is required', 1;

    EXEC alg.usp_rebuild_space_topology @warehouse_id = @warehouse_id,
                                        @snapshot_generation = @generation;
    EXEC alg.usp_sync_inventory_snapshot @warehouse_id = @warehouse_id,
                                         @snapshot_generation = @generation;
    EXEC alg.usp_rebuild_lane_state @warehouse_id = @warehouse_id,
                                    @snapshot_generation = @generation;
END;
GO

CREATE PROCEDURE alg.usp_claim_shadow_instance
    @warehouse_id varchar(80),
    @instance_id varchar(80),
    @claimed bit OUTPUT
AS
BEGIN
    SET NOCOUNT ON;
    SET XACT_ABORT ON;
    SET @claimed = 0;

    IF NULLIF(LTRIM(RTRIM(@warehouse_id)), '') IS NULL
        THROW 51004, 'warehouse_id is required', 1;
    IF NULLIF(LTRIM(RTRIM(@instance_id)), '') IS NULL
        THROW 51005, 'instance_id is required', 1;

    DECLARE @now datetime2 = SYSUTCDATETIME(), @current_instance varchar(80),
            @last_seen_at datetime2;

    BEGIN TRY
        BEGIN TRANSACTION;
        SELECT @current_instance = heartbeat.instance_id, @last_seen_at = heartbeat.last_seen_at
        FROM alg.shadow_instance_heartbeat heartbeat WITH (UPDLOCK, HOLDLOCK)
        WHERE heartbeat.warehouse_id = @warehouse_id;

        IF @current_instance IS NULL
        BEGIN
            INSERT INTO alg.shadow_instance_heartbeat (warehouse_id, instance_id, claimed_at, last_seen_at)
            VALUES (@warehouse_id, @instance_id, @now, @now);
            SET @claimed = 1;
        END
        ELSE IF @current_instance = @instance_id
        BEGIN
            UPDATE alg.shadow_instance_heartbeat
            SET last_seen_at = @now
            WHERE warehouse_id = @warehouse_id;
            SET @claimed = 1;
        END
        ELSE IF @last_seen_at < DATEADD(second, -120, @now)
        BEGIN
            UPDATE alg.shadow_instance_heartbeat
            SET instance_id = @instance_id, claimed_at = @now, last_seen_at = @now
            WHERE warehouse_id = @warehouse_id;
            SET @claimed = 1;
        END;

        COMMIT TRANSACTION;
    END TRY
    BEGIN CATCH
        IF @@TRANCOUNT > 0
            ROLLBACK TRANSACTION;
        THROW;
    END CATCH;
    SELECT @claimed AS claimed;
END;
GO

CREATE PROCEDURE alg.usp_refresh_shadow_snapshot
    @warehouse_id varchar(80),
    @snapshot_generation bigint,
    @instance_id varchar(80)
AS
BEGIN
    SET NOCOUNT ON;
    SET XACT_ABORT ON;

    IF NULLIF(LTRIM(RTRIM(@warehouse_id)), '') IS NULL
        THROW 51006, 'warehouse_id is required', 1;
    IF NULLIF(LTRIM(RTRIM(@instance_id)), '') IS NULL
        THROW 51007, 'instance_id is required', 1;
    IF @snapshot_generation <= 0
        THROW 51008, 'snapshot_generation must be positive', 1;

    DECLARE @pre_source_count bigint, @post_source_count bigint,
            @pre_source_signature varbinary(32), @post_source_signature varbinary(32),
            @source_payload nvarchar(max), @synchronized_at datetime2,
            @numeric_warehouse_id decimal(38,0),
            @current_generation bigint;

    IF LTRIM(RTRIM(@warehouse_id)) NOT LIKE '%[^0-9]%'
        SET @numeric_warehouse_id = TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(@warehouse_id)));

    BEGIN TRY
        BEGIN TRANSACTION;

        SELECT @current_generation = state.snapshot_generation
        FROM alg.shadow_snapshot_state state WITH (UPDLOCK, HOLDLOCK)
        WHERE state.warehouse_id = @warehouse_id;

        IF @current_generation IS NOT NULL AND @snapshot_generation <= @current_generation
            THROW 51009, 'snapshot_generation must advance the published generation', 1;

        SELECT @pre_source_count =
            (SELECT COUNT_BIG(*) FROM dbo.bas_cargo_space cargo WITH (NOLOCK) WHERE cargo.warehouse_id = @warehouse_id)
          + (SELECT COUNT_BIG(*) FROM dbo.bas_stock stock WITH (NOLOCK)
             JOIN dbo.bas_cargo_space cargo WITH (NOLOCK) ON cargo.cargo_space_id = stock.cargo_space_id
             WHERE cargo.warehouse_id = @warehouse_id)
          + (SELECT COUNT_BIG(*) FROM dbo.mydo_job job WITH (NOLOCK) WHERE job.warehouse_id = @warehouse_id)
          + (SELECT COUNT_BIG(*) FROM dbo.mydo_job_detail detail WITH (NOLOCK)
             JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
             WHERE job.warehouse_id = @warehouse_id)
	          + (SELECT COUNT_BIG(*) FROM dbo.in_stockin_detail inbound WITH (NOLOCK)
	             JOIN dbo.mydo_job_detail detail WITH (NOLOCK) ON detail.stockin_detail_id = inbound.stockin_detail_id
	             JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
	             JOIN dbo.in_stockin stockin WITH (NOLOCK) ON stockin.stockin_id = inbound.stockin_id
	             WHERE (job.finish_time IS NULL OR LTRIM(RTRIM(job.finish_time)) = '')
	               AND (LTRIM(RTRIM(stockin.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
	                    OR (@numeric_warehouse_id IS NOT NULL
	                        AND LTRIM(RTRIM(stockin.warehouse_id)) NOT LIKE '%[^0-9]%'
	                        AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockin.warehouse_id))) = @numeric_warehouse_id)))
	          + (SELECT COUNT_BIG(*) FROM dbo.out_stockout_detail outbound WITH (NOLOCK)
	             JOIN dbo.out_stockout stockout WITH (NOLOCK) ON stockout.stockout_id = outbound.stockout_id
	             WHERE ISNULL(outbound.quantity, 0) > ISNULL(outbound.allot_quantity, 0)
	               AND (LTRIM(RTRIM(stockout.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
	                    OR (@numeric_warehouse_id IS NOT NULL
	                        AND LTRIM(RTRIM(stockout.warehouse_id)) NOT LIKE '%[^0-9]%'
	                        AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockout.warehouse_id))) = @numeric_warehouse_id)));

        SELECT @source_payload = ISNULL(CONVERT(nvarchar(max), (
            SELECT cargo.cargo_space_id AS [@space_id], cargo.warehouse_id AS [@warehouse_id],
                   cargo.group_coding_no AS [@lane_code], cargo.cargo_area_id AS [@cargo_area_id],
                   cargo.roadway_id AS [@roadway_id], cargo.cargo_space_name AS [@display_name],
                   cargo.the_platoon AS [@platoon], cargo.the_column AS [@column_no],
                   cargo.the_floor AS [@floor_no], cargo.group_coding AS [@depth_raw],
                   cargo.status AS [@space_status], cargo.locked AS [@locked], cargo.tray_code AS [@tray_code]
            FROM dbo.bas_cargo_space cargo WITH (NOLOCK)
            WHERE cargo.warehouse_id = @warehouse_id
            ORDER BY cargo.cargo_space_id
            FOR XML PATH('cargo'), ROOT('cargo_spaces'), TYPE
        )), '');

        SET @source_payload = @source_payload + '|STOCK|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT stock.stock_id AS [@stock_id], stock.cargo_space_id AS [@space_id],
                   stock.warehouse_id AS [@warehouse_id], stock.tray_code AS [@tray_code],
                   stock.spare4 AS [@project_no], stock.product_id AS [@product_no],
                   stock.lot_no AS [@lot_no], stock.in_date AS [@in_date],
                   stock.quantity AS [@quantity], stock.out_quantity AS [@out_quantity]
            FROM dbo.bas_stock stock WITH (NOLOCK)
            JOIN dbo.bas_cargo_space cargo WITH (NOLOCK) ON cargo.cargo_space_id = stock.cargo_space_id
            WHERE cargo.warehouse_id = @warehouse_id
            ORDER BY stock.stock_id
            FOR XML PATH('stock'), ROOT('stocks'), TYPE
        )), '');

        SET @source_payload = @source_payload + '|JOB|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT job.job_id AS [@job_id], job.status AS [@status], job.task_no AS [@task_no],
                   job.tray_code AS [@tray_code], job.cargo_space_id AS [@source_space_id],
                   job.target_cargo_space_id AS [@target_space_id], job.in_or_out AS [@in_or_out],
                   job.create_time AS [@create_time], job.finish_time AS [@finish_time],
                   job.warehouse_id AS [@warehouse_id]
            FROM dbo.mydo_job job WITH (NOLOCK)
            WHERE job.warehouse_id = @warehouse_id
            ORDER BY job.job_id
            FOR XML PATH('job'), ROOT('jobs'), TYPE
        )), '');

        SET @source_payload = @source_payload + '|JOB_DETAIL|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT detail.job_detail_id AS [@job_detail_id], detail.job_id AS [@job_id],
                   detail.stock_id AS [@stock_id], detail.lot_no AS [@lot_no],
                   detail.in_date AS [@in_date], detail.in_or_out_id AS [@request_id],
                   detail.in_or_out_d_id AS [@request_detail_id], detail.product_id AS [@product_id],
                   detail.stock_status AS [@stock_status], detail.in_quantity AS [@in_quantity],
                   detail.stock_quantity AS [@stock_quantity], detail.out_quantity AS [@out_quantity],
                   detail.pick_quantity AS [@pick_quantity], detail.stockin_detail_id AS [@stockin_detail_id]
            FROM dbo.mydo_job_detail detail WITH (NOLOCK)
            JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
            WHERE job.warehouse_id = @warehouse_id
            ORDER BY detail.job_detail_id
            FOR XML PATH('job_detail'), ROOT('job_details'), TYPE
        )), '');

        SET @source_payload = @source_payload + '|INBOUND|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT inbound.stockin_detail_id AS [@stockin_detail_id], inbound.stockin_guid AS [@stockin_guid],
                   inbound.stockin_id AS [@stockin_id], inbound.lot_no AS [@lot_no],
                   inbound.product_id AS [@product_id], inbound.quantity AS [@quantity],
                   inbound.assign_num AS [@assign_num], inbound.spare4 AS [@project_no],
                   job.warehouse_id AS [@warehouse_id], stockin.warehouse_id AS [@source_warehouse_id],
                   job.task_no AS [@task_no],
                   job.tray_code AS [@tray_code], detail.job_id AS [@job_id],
                   detail.in_or_out_id AS [@request_id]
            FROM dbo.in_stockin_detail inbound WITH (NOLOCK)
            JOIN dbo.mydo_job_detail detail WITH (NOLOCK) ON detail.stockin_detail_id = inbound.stockin_detail_id
            JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
            JOIN dbo.in_stockin stockin WITH (NOLOCK) ON stockin.stockin_id = inbound.stockin_id
            WHERE (job.finish_time IS NULL OR LTRIM(RTRIM(job.finish_time)) = '')
              AND (LTRIM(RTRIM(stockin.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
                   OR (@numeric_warehouse_id IS NOT NULL
                       AND LTRIM(RTRIM(stockin.warehouse_id)) NOT LIKE '%[^0-9]%'
                       AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockin.warehouse_id))) = @numeric_warehouse_id))
            ORDER BY inbound.stockin_detail_id
            FOR XML PATH('inbound'), ROOT('inbound_requests'), TYPE
        )), '');

        SET @source_payload = @source_payload + '|OUTBOUND|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT outbound.stockout_detail_id AS [@stockout_detail_id], outbound.warehouse_id AS [@warehouse_id],
                   stockout.warehouse_id AS [@source_warehouse_id],
                   warehouse.project_no AS [@project_no],
                   outbound.product_id AS [@product_id], outbound.lot_no AS [@lot_no],
                   outbound.quantity AS [@quantity], outbound.assign_num AS [@assign_num],
                   outbound.allot_quantity AS [@allot_quantity]
            FROM dbo.out_stockout_detail outbound WITH (NOLOCK)
            JOIN dbo.out_stockout stockout WITH (NOLOCK) ON stockout.stockout_id = outbound.stockout_id
            LEFT JOIN dbo.bas_warehouse warehouse WITH (NOLOCK)
	          ON warehouse.warehouse_id = @warehouse_id
            WHERE ISNULL(outbound.quantity, 0) > ISNULL(outbound.allot_quantity, 0)
              AND (LTRIM(RTRIM(stockout.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
                   OR (@numeric_warehouse_id IS NOT NULL
                       AND LTRIM(RTRIM(stockout.warehouse_id)) NOT LIKE '%[^0-9]%'
                       AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockout.warehouse_id))) = @numeric_warehouse_id))
            ORDER BY outbound.stockout_detail_id
            FOR XML PATH('outbound'), ROOT('outbound_requests'), TYPE
        )), '');

        SET @pre_source_signature = HASHBYTES('SHA2_256', CONVERT(varbinary(max), @source_payload));

        DELETE FROM alg.cargo_space_snapshot WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.inventory_item_snapshot WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.active_job_snapshot WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.shadow_inbound_request_snapshot WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.shadow_outbound_request_snapshot WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.space_topology WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.space_inventory_summary WHERE warehouse_id = @warehouse_id;
        DELETE FROM alg.lane_state WHERE warehouse_id = @warehouse_id;

        SET @synchronized_at = SYSUTCDATETIME();

        INSERT INTO alg.cargo_space_snapshot (
            warehouse_id, space_id, lane_code, cargo_area_id, roadway_id, display_name,
            platoon, column_no, floor_no, depth_raw, space_status, locked, tray_code,
            snapshot_generation, synchronized_at)
        SELECT cargo.warehouse_id, cargo.cargo_space_id, cargo.group_coding_no,
               cargo.cargo_area_id, cargo.roadway_id, cargo.cargo_space_name,
               cargo.the_platoon, cargo.the_column, cargo.the_floor, cargo.group_coding,
               cargo.status, cargo.locked, cargo.tray_code,
               @snapshot_generation, @synchronized_at
        FROM dbo.bas_cargo_space cargo WITH (NOLOCK)
        WHERE cargo.warehouse_id = @warehouse_id;

        INSERT INTO alg.inventory_item_snapshot (
            warehouse_id, stock_id, space_id, tray_code, project_no, product_no, lot_no,
            inbound_time_raw, inbound_at, quantity, out_quantity, snapshot_generation, synchronized_at)
        SELECT cargo.warehouse_id, stock.stock_id, stock.cargo_space_id, stock.tray_code,
               stock.spare4, stock.product_id, stock.lot_no, stock.in_date,
               TRY_CONVERT(datetime2, stock.in_date), stock.quantity, stock.out_quantity,
               @snapshot_generation, @synchronized_at
        FROM dbo.bas_stock stock WITH (NOLOCK)
        JOIN dbo.bas_cargo_space cargo WITH (NOLOCK) ON cargo.cargo_space_id = stock.cargo_space_id
        WHERE cargo.warehouse_id = @warehouse_id;

        INSERT INTO alg.active_job_snapshot (
            warehouse_id, job_id, task_no, tray_code, source_space_id, target_space_id,
            status, in_or_out, create_time_raw, snapshot_generation, synchronized_at)
        SELECT job.warehouse_id, job.job_id, job.task_no, job.tray_code,
               NULLIF(LTRIM(RTRIM(job.cargo_space_id)), ''),
               NULLIF(LTRIM(RTRIM(job.target_cargo_space_id)), ''), job.status,
               job.in_or_out, job.create_time, @snapshot_generation, @synchronized_at
        FROM dbo.mydo_job job WITH (NOLOCK)
        WHERE job.warehouse_id = @warehouse_id
          AND (job.finish_time IS NULL OR LTRIM(RTRIM(job.finish_time)) = '');

        ;WITH inbound_source AS (
            SELECT @warehouse_id AS warehouse_id, job.task_no, job.tray_code,
                   detail.in_or_out_id AS request_id, job.warehouse_id AS raw_warehouse_id,
                   inbound.spare4 AS project_no, inbound.product_id, inbound.lot_no,
                   ROW_NUMBER() OVER (
                       PARTITION BY job.task_no, job.tray_code
                       ORDER BY inbound.stockin_detail_id, detail.job_detail_id) AS source_rank
            FROM dbo.in_stockin_detail inbound WITH (NOLOCK)
            JOIN dbo.mydo_job_detail detail WITH (NOLOCK)
              ON detail.stockin_detail_id = inbound.stockin_detail_id
            JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
            JOIN dbo.in_stockin stockin WITH (NOLOCK) ON stockin.stockin_id = inbound.stockin_id
            WHERE (job.finish_time IS NULL OR LTRIM(RTRIM(job.finish_time)) = '')
              AND (LTRIM(RTRIM(stockin.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
                   OR (@numeric_warehouse_id IS NOT NULL
                       AND LTRIM(RTRIM(stockin.warehouse_id)) NOT LIKE '%[^0-9]%'
                       AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockin.warehouse_id))) = @numeric_warehouse_id))
        )
        INSERT INTO alg.shadow_inbound_request_snapshot (
            warehouse_id, task_no, tray_no, request_id, raw_warehouse_id,
            project_no, product_id, lot_no, snapshot_generation, synchronized_at)
        SELECT @warehouse_id, ISNULL(task_no, ''), ISNULL(tray_code, ''), request_id,
               raw_warehouse_id, project_no, product_id, lot_no,
               @snapshot_generation, @synchronized_at
        FROM inbound_source
        WHERE source_rank = 1;

        INSERT INTO alg.shadow_outbound_request_snapshot (
            warehouse_id, stockout_detail_id, raw_warehouse_id, project_no, product_id,
            requested_quantity, snapshot_generation, synchronized_at)
        SELECT @warehouse_id, outbound.stockout_detail_id, outbound.warehouse_id,
               NULLIF(LTRIM(RTRIM(warehouse.project_no)), ''), outbound.product_id,
               CONVERT(decimal(38,6), outbound.quantity),
               @snapshot_generation, @synchronized_at
        FROM dbo.out_stockout_detail outbound WITH (NOLOCK)
        JOIN dbo.out_stockout stockout WITH (NOLOCK) ON stockout.stockout_id = outbound.stockout_id
        LEFT JOIN dbo.bas_warehouse warehouse WITH (NOLOCK)
	      ON warehouse.warehouse_id = @warehouse_id
	    WHERE ISNULL(outbound.quantity, 0) > ISNULL(outbound.allot_quantity, 0)
	      AND (LTRIM(RTRIM(stockout.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
	           OR (@numeric_warehouse_id IS NOT NULL
	               AND LTRIM(RTRIM(stockout.warehouse_id)) NOT LIKE '%[^0-9]%'
	               AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockout.warehouse_id))) = @numeric_warehouse_id));

        EXEC alg.usp_rebuild_space_topology @warehouse_id = @warehouse_id,
                                            @snapshot_generation = @snapshot_generation;
        EXEC alg.usp_sync_inventory_snapshot @warehouse_id = @warehouse_id,
                                             @snapshot_generation = @snapshot_generation;
        EXEC alg.usp_rebuild_lane_state @warehouse_id = @warehouse_id,
                                        @snapshot_generation = @snapshot_generation;

        SELECT @post_source_count =
            (SELECT COUNT_BIG(*) FROM dbo.bas_cargo_space cargo WITH (NOLOCK) WHERE cargo.warehouse_id = @warehouse_id)
          + (SELECT COUNT_BIG(*) FROM dbo.bas_stock stock WITH (NOLOCK)
             JOIN dbo.bas_cargo_space cargo WITH (NOLOCK) ON cargo.cargo_space_id = stock.cargo_space_id
             WHERE cargo.warehouse_id = @warehouse_id)
          + (SELECT COUNT_BIG(*) FROM dbo.mydo_job job WITH (NOLOCK) WHERE job.warehouse_id = @warehouse_id)
          + (SELECT COUNT_BIG(*) FROM dbo.mydo_job_detail detail WITH (NOLOCK)
             JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
             WHERE job.warehouse_id = @warehouse_id)
	          + (SELECT COUNT_BIG(*) FROM dbo.in_stockin_detail inbound WITH (NOLOCK)
	             JOIN dbo.mydo_job_detail detail WITH (NOLOCK) ON detail.stockin_detail_id = inbound.stockin_detail_id
	             JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
	             JOIN dbo.in_stockin stockin WITH (NOLOCK) ON stockin.stockin_id = inbound.stockin_id
	             WHERE (job.finish_time IS NULL OR LTRIM(RTRIM(job.finish_time)) = '')
	               AND (LTRIM(RTRIM(stockin.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
	                    OR (@numeric_warehouse_id IS NOT NULL
	                        AND LTRIM(RTRIM(stockin.warehouse_id)) NOT LIKE '%[^0-9]%'
	                        AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockin.warehouse_id))) = @numeric_warehouse_id)))
	          + (SELECT COUNT_BIG(*) FROM dbo.out_stockout_detail outbound WITH (NOLOCK)
	             JOIN dbo.out_stockout stockout WITH (NOLOCK) ON stockout.stockout_id = outbound.stockout_id
	             WHERE ISNULL(outbound.quantity, 0) > ISNULL(outbound.allot_quantity, 0)
	               AND (LTRIM(RTRIM(stockout.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
	                    OR (@numeric_warehouse_id IS NOT NULL
	                        AND LTRIM(RTRIM(stockout.warehouse_id)) NOT LIKE '%[^0-9]%'
	                        AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockout.warehouse_id))) = @numeric_warehouse_id)));

        SELECT @source_payload = ISNULL(CONVERT(nvarchar(max), (
            SELECT cargo.cargo_space_id AS [@space_id], cargo.warehouse_id AS [@warehouse_id],
                   cargo.group_coding_no AS [@lane_code], cargo.cargo_area_id AS [@cargo_area_id],
                   cargo.roadway_id AS [@roadway_id], cargo.cargo_space_name AS [@display_name],
                   cargo.the_platoon AS [@platoon], cargo.the_column AS [@column_no],
                   cargo.the_floor AS [@floor_no], cargo.group_coding AS [@depth_raw],
                   cargo.status AS [@space_status], cargo.locked AS [@locked], cargo.tray_code AS [@tray_code]
            FROM dbo.bas_cargo_space cargo WITH (NOLOCK)
            WHERE cargo.warehouse_id = @warehouse_id
            ORDER BY cargo.cargo_space_id
            FOR XML PATH('cargo'), ROOT('cargo_spaces'), TYPE
        )), '');
        SET @source_payload = @source_payload + '|STOCK|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT stock.stock_id AS [@stock_id], stock.cargo_space_id AS [@space_id],
                   stock.warehouse_id AS [@warehouse_id], stock.tray_code AS [@tray_code],
                   stock.spare4 AS [@project_no], stock.product_id AS [@product_no],
                   stock.lot_no AS [@lot_no], stock.in_date AS [@in_date],
                   stock.quantity AS [@quantity], stock.out_quantity AS [@out_quantity]
            FROM dbo.bas_stock stock WITH (NOLOCK)
            JOIN dbo.bas_cargo_space cargo WITH (NOLOCK) ON cargo.cargo_space_id = stock.cargo_space_id
            WHERE cargo.warehouse_id = @warehouse_id
            ORDER BY stock.stock_id
            FOR XML PATH('stock'), ROOT('stocks'), TYPE
        )), '');
        SET @source_payload = @source_payload + '|JOB|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT job.job_id AS [@job_id], job.status AS [@status], job.task_no AS [@task_no],
                   job.tray_code AS [@tray_code], job.cargo_space_id AS [@source_space_id],
                   job.target_cargo_space_id AS [@target_space_id], job.in_or_out AS [@in_or_out],
                   job.create_time AS [@create_time], job.finish_time AS [@finish_time],
                   job.warehouse_id AS [@warehouse_id]
            FROM dbo.mydo_job job WITH (NOLOCK)
            WHERE job.warehouse_id = @warehouse_id
            ORDER BY job.job_id
            FOR XML PATH('job'), ROOT('jobs'), TYPE
        )), '');
        SET @source_payload = @source_payload + '|JOB_DETAIL|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT detail.job_detail_id AS [@job_detail_id], detail.job_id AS [@job_id],
                   detail.stock_id AS [@stock_id], detail.lot_no AS [@lot_no],
                   detail.in_date AS [@in_date], detail.in_or_out_id AS [@request_id],
                   detail.in_or_out_d_id AS [@request_detail_id], detail.product_id AS [@product_id],
                   detail.stock_status AS [@stock_status], detail.in_quantity AS [@in_quantity],
                   detail.stock_quantity AS [@stock_quantity], detail.out_quantity AS [@out_quantity],
                   detail.pick_quantity AS [@pick_quantity], detail.stockin_detail_id AS [@stockin_detail_id]
            FROM dbo.mydo_job_detail detail WITH (NOLOCK)
            JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
            WHERE job.warehouse_id = @warehouse_id
            ORDER BY detail.job_detail_id
            FOR XML PATH('job_detail'), ROOT('job_details'), TYPE
        )), '');
        SET @source_payload = @source_payload + '|INBOUND|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT inbound.stockin_detail_id AS [@stockin_detail_id], inbound.stockin_guid AS [@stockin_guid],
                   inbound.stockin_id AS [@stockin_id], inbound.lot_no AS [@lot_no],
                   inbound.product_id AS [@product_id], inbound.quantity AS [@quantity],
                   inbound.assign_num AS [@assign_num], inbound.spare4 AS [@project_no],
                   job.warehouse_id AS [@warehouse_id], stockin.warehouse_id AS [@source_warehouse_id],
                   job.task_no AS [@task_no],
                   job.tray_code AS [@tray_code], detail.job_id AS [@job_id],
                   detail.in_or_out_id AS [@request_id]
            FROM dbo.in_stockin_detail inbound WITH (NOLOCK)
            JOIN dbo.mydo_job_detail detail WITH (NOLOCK) ON detail.stockin_detail_id = inbound.stockin_detail_id
            JOIN dbo.mydo_job job WITH (NOLOCK) ON job.job_id = detail.job_id
            JOIN dbo.in_stockin stockin WITH (NOLOCK) ON stockin.stockin_id = inbound.stockin_id
            WHERE (job.finish_time IS NULL OR LTRIM(RTRIM(job.finish_time)) = '')
              AND (LTRIM(RTRIM(stockin.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
                   OR (@numeric_warehouse_id IS NOT NULL
                       AND LTRIM(RTRIM(stockin.warehouse_id)) NOT LIKE '%[^0-9]%'
                       AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockin.warehouse_id))) = @numeric_warehouse_id))
            ORDER BY inbound.stockin_detail_id
            FOR XML PATH('inbound'), ROOT('inbound_requests'), TYPE
        )), '');
        SET @source_payload = @source_payload + '|OUTBOUND|' + ISNULL(CONVERT(nvarchar(max), (
            SELECT outbound.stockout_detail_id AS [@stockout_detail_id], outbound.warehouse_id AS [@warehouse_id],
                   stockout.warehouse_id AS [@source_warehouse_id],
                   warehouse.project_no AS [@project_no],
                   outbound.product_id AS [@product_id], outbound.lot_no AS [@lot_no],
                   outbound.quantity AS [@quantity], outbound.assign_num AS [@assign_num],
                   outbound.allot_quantity AS [@allot_quantity]
            FROM dbo.out_stockout_detail outbound WITH (NOLOCK)
            JOIN dbo.out_stockout stockout WITH (NOLOCK) ON stockout.stockout_id = outbound.stockout_id
            LEFT JOIN dbo.bas_warehouse warehouse WITH (NOLOCK)
	          ON warehouse.warehouse_id = @warehouse_id
            WHERE ISNULL(outbound.quantity, 0) > ISNULL(outbound.allot_quantity, 0)
              AND (LTRIM(RTRIM(stockout.warehouse_id)) = LTRIM(RTRIM(@warehouse_id))
                   OR (@numeric_warehouse_id IS NOT NULL
                       AND LTRIM(RTRIM(stockout.warehouse_id)) NOT LIKE '%[^0-9]%'
                       AND TRY_CONVERT(decimal(38,0), LTRIM(RTRIM(stockout.warehouse_id))) = @numeric_warehouse_id))
            ORDER BY outbound.stockout_detail_id
            FOR XML PATH('outbound'), ROOT('outbound_requests'), TYPE
        )), '');
        SET @post_source_signature = HASHBYTES('SHA2_256', CONVERT(varbinary(max), @source_payload));

        IF @pre_source_count <> @post_source_count
           OR @pre_source_signature <> @post_source_signature
            THROW 51025, 'FT546 shadow refresh refused: source changed during refresh', 1;

        MERGE alg.shadow_snapshot_state WITH (HOLDLOCK) AS target
        USING (SELECT @warehouse_id AS warehouse_id) AS source
        ON source.warehouse_id = target.warehouse_id
        WHEN MATCHED THEN UPDATE SET
            snapshot_generation = @snapshot_generation,
            snapshot_status = 'PUBLISHED',
            synchronized_at = SYSUTCDATETIME(),
            source_signature = @post_source_signature,
            source_row_count = @post_source_count,
            instance_id = @instance_id,
            last_error = NULL
        WHEN NOT MATCHED THEN INSERT (
            warehouse_id, snapshot_generation, snapshot_status, synchronized_at,
            source_signature, source_row_count, instance_id, last_error)
        VALUES (
            @warehouse_id, @snapshot_generation, 'PUBLISHED', SYSUTCDATETIME(),
            @post_source_signature, @post_source_count, @instance_id, NULL);

        COMMIT TRANSACTION;
    END TRY
    BEGIN CATCH
        IF @@TRANCOUNT > 0
            ROLLBACK TRANSACTION;
        THROW;
    END CATCH;
END;
GO