# Brink Labor GetShifts Staging

## Goal
Add BrinkPOS Labor `GetShifts` into the ETL flow as raw staging data for Redshift.

This staging layer is **1:1 with the Brink SOAP payload**, plus technical properties required for ingestion, traceability, and idempotent load handling.

## Source
- Endpoint: `https://api2.brinkpos.net/Labor2.svc`
- SOAP Action: `http://www.brinksoftware.com/webservices/labor/v2/ILaborWebService2/GetShifts`
- Docs: `https://brinkapiportal.parpos.com/labor2#GetShifts`

## Scope
Add two staging tables / parquet artifacts:
1. `staging.brink_labor_shifts`
2. `staging.brink_labor_breaks`

## Rules
- Staging is raw, not curated.
- Business columns follow current Brink staging convention: Brink field names flattened to `snake_case`.
- Parquet row models and staging row models are the same models.
- Add the standard `_etl_*` metadata columns from `NormalizedRowBase` + `BrinkStagingRowBase`.
- Do not add legacy generic metadata columns like `source_system`, `source_endpoint`, `payload_json`, `record_hash`, or `loaded_at_utc` unless current framework requires them.
- Breaks are emitted as a separate child artifact/table, linked to the parent via `shift_id`.
- Do not invent raw business keys that Brink does not send. `Break` has no Brink `Id`, so `staging.brink_labor_breaks` stays linked by `shift_id` plus raw break fields and ETL metadata.
- This must plug into the existing `extract_raw -> normalize_load -> load_to_staging` Brink ETL job, not a separate one-off flow.

## Flow
1. Extract from Brink POS
   - `BrinkPosExtractOrchestrator` starts the job.
   - `BrinkPosEndpointExecutor` calls Brink SOAP endpoints.
   - Main source is `ILaborWebService2/GetShifts`.
2. Store raw response
    - Raw Brink XML is saved to S3 via `IEtlS3Storage.WriteRawAsync(...)`.
    - Extract stage details must be persisted as a `StageDiagnosticsEnvelope<T>` payload, not plain JSON, because downstream normalize/load uses `StageDiagnosticsJson.DeserializeDetailsWithLegacyFallback<T>(...)`.
3. Normalize / map data
   - `BrinkPosNormalizeLoadJobs` runs the normalization step.
   - Current repo pattern for staging uses `BrinkMultiArtifactProfileBase` + `IBrinkXmlEnvelopeParser` to read the Brink SOAP payload.
   - `GetShifts` should follow that same staging-profile pattern unless a shared parser/mapper abstraction is introduced as part of the implementation.
4. Write Parquet
   - `ParquetBatchWriter<T>` writes staging row models directly into Parquet files.
   - Parent and child collections are written as separate artifacts.
   - Files are compressed and uploaded for loading.
5. Load to Redshift staging
   - `BrinkPosLoadToStagingJobs` triggers the load.
   - `RedshiftStagingLoadExecutor` runs `COPY` from S3 into Redshift staging.
   - It deletes existing rows using current staging catalog idempotency predicates before `COPY`.

## Source Contract

### Shift data
| Data Name | Data Type | Data Description |
|---|---|---|
| Breaks | Array of Break | Breaks for this Shift |
| BusinessDate | DateTime | Business date of this Shift (the time component will always return as 00:00:00) |
| ClockOutBusinessDate | DateTime | Business date on which this Shift was clocked out (the time component will always return as 00:00:00) |
| DeclaredTips | decimal | Tips declared for this Shift |
| EmployeeId | int | Maps to the Id of the Employee for this Shift |
| EndTime | DateTimeOffset | End time of this Shift |
| ExtendedMinutesWorked | int | Amount of minutes worked for this Shift outside of regular working hours, to be paid at a regular pay rate |
| Id | Guid | Unique Id of this Shift |
| JobId | int | Maps to the Id of the Job for this Shift |
| MinutesWorked | int | Amount of total minutes worked for this Shift |
| ModifiedTime | DateTimeOffset | Date and time of the last time this Shift was modified (e.g. by a clock in/out or break event) |
| Number | byte | Incremental count of the employee's Shifts for a particular business date (e.g. returns `1` for the first shift of the day, `2` for the second, etc.) |
| OvertimeMinutesWorked | int | Amount of minutes worked for this Shift in overtime, to be paid at an overtime pay rate |
| PayRate | decimal | Pay rate that was current for the job at the time of this Shift |
| RegularMinutesWorked | int | Amount of minutes worked for this Shift within regular working hours, to be paid at a regular pay rate |
| StartTime | DateTimeOffset | Start time of this Shift |

### Break data
| Data Name | Data Type | Data Description |
|---|---|---|
| EndTime | DateTimeOffset | End time of this Break |
| Minutes | int | Minutes of this paid Break (unpaid Breaks are not included) |
| Number | byte | Incremental count of the employee's Breaks for a particular business date (e.g. returns `1` for the first break of the day, `2` for the second, etc.) |
| IsPaid | bool | `true` if this is a paid Break, `false` otherwise |
| StartTime | DateTimeOffset | Start time of this Break |

## Current Pattern Validation

### Best analogous implementation
- `Sales2GetOrdersStagingProfile`
- `BrinkSales2GetOrdersRow`
- `BrinkSales2OrderEntriesRow`
- `BrinkPosStagingTableCatalog`
- `backend/Kds.Ai.Warehouse/Sql/Bootstrap/Tenant/Staging/BrinkPos/0605_create_staging_brink_sales2_get_orders.sql`

### Confirmed current pattern
- Brink staging rows inherit from `BrinkStagingRowBase`.
- `BrinkStagingRowBase` extends `NormalizedRowBase`.
- Parquet models are the staging row models; there is no separate parquet-only schema.
- Nested collections are flattened into separate artifacts/tables.
- Business columns use `snake_case`.
- Technical metadata is standardized through `_etl_*` columns.

### Verified current repo constraints that this integration must remove
These were the constraints discovered at the start of the work and are exactly the kinds of repo couplings that must be checked when adding any new Brink contract:
- `BrinkPosEndpointCatalog` had been orders-only.
- `BrinkPosEndpointExecutor` had been orders-only.
- `BrinkPosNormalizationContextFactory` had been orders-centric.
- `BrinkPosNormalizationCatalog` and `BrinkPosStagingTableCatalog` had been orders-only catalogs.
- `BrinkPosNormalizeLoadJobs` / `BrinkPosLoadToStagingJobs` had assumed a single orders contract group.
- DI registration had only active orders staging profile(s).
- Raw capture support, typed facade support, and live-test coverage had to be added separately.

## Implemented Repo Pattern For New Contracts

The final `GetShifts` implementation established the repo pattern that future Brink contract onboarding should follow:

1. Transport / endpoint support
   - Add service/operation enum values.
   - Add SOAP envelope builder support.
   - Add raw request builder support.
   - Add raw capture method to `IBrinkRawClient` / `BrinkRawClient`.
   - Add endpoint definition to `BrinkPosEndpointCatalog`.
   - Update `BrinkPosEndpointExecutor` to dispatch the new endpoint.

2. Operation metadata
   - Add the contract group to `BrinkPosOperationCatalog`.
   - Define:
     - `stream_name`
     - `operation_name`
     - `target_contract_group`
     - `parser_version`
     - request metadata
     - contract list
     - staging table list

3. Extract details contract
   - Ensure extract writes endpoint-level details into `BrinkPosExtractBatchDetails`.
   - Persist those details wrapped in `StageDiagnosticsEnvelope<BrinkPosExtractBatchDetails>`.
   - This is required so `normalize_load` can process multiple raw Brink artifacts safely.

4. Normalization / staging
   - Add staging row models inheriting from `BrinkStagingRowBase`.
   - Add a dedicated staging profile using `BrinkMultiArtifactProfileBase` when the payload has child collections.
   - Register the profile in `AddBrinkPosEtlIntegration()`.
   - Add the contract group to `BrinkPosNormalizationCatalog`.

5. Load to Redshift staging
   - Add staging tables and delete predicates to `BrinkPosStagingTableCatalog`.
   - Ensure `BrinkPosLoadToStagingJobs` resolves the table set from the normalized contracts, not from hardcoded orders-only assumptions.

6. Optional typed facade support
   - If the endpoint should be available through `IBrinkPosFacade`, add:
     - request model if needed
     - typed response model
     - service client (for example `LaborClient`)
     - facade interface method
     - facade implementation
     - DI registration

7. Validation coverage
   - Add unit transport coverage.
   - Add staging-profile coverage.
   - Add ETL catalog / executor coverage.
   - Add live ETL coverage when applicable.
   - Add live facade/raw coverage when XML artifacts in `temp/` are part of the expected workflow.

## Table 1: `staging.brink_labor_shifts`

### Purpose
Raw shift-level staging table. One row per Brink shift.

### Business columns
| Column name | Type | Notes |
|---|---|---|
| business_date | timestamp | Brink `BusinessDate`; business-day date value stored using current staging date pattern |
| clock_out_business_date | timestamp | Brink `ClockOutBusinessDate` |
| declared_tips | decimal(18,4) | Brink `DeclaredTips` |
| employee_id | bigint | Brink `EmployeeId` |
| end_time | timestamptz | Brink `EndTime` |
| extended_minutes_worked | int | Brink `ExtendedMinutesWorked` |
| shift_id | varchar(36) | Brink `Id` GUID; parent key for child table |
| job_id | bigint | Brink `JobId` |
| minutes_worked | int | Brink `MinutesWorked` |
| modified_time | timestamptz | Brink `ModifiedTime` |
| number | smallint | Brink `Number` |
| overtime_minutes_worked | int | Brink `OvertimeMinutesWorked` |
| pay_rate | decimal(18,4) | Brink `PayRate` |
| regular_minutes_worked | int | Brink `RegularMinutesWorked` |
| start_time | timestamptz | Brink `StartTime` |

### Technical columns
| Column name | Type | Notes |
|---|---|---|
| _etl_batch_id | varchar | Standard normalized metadata |
| _etl_company_id | varchar | Standard normalized metadata |
| _etl_tenant_id | varchar | Tenant identifier used by staging load |
| _etl_source_type | varchar | Standard normalized metadata |
| _etl_stream_name | varchar | Brink stream/artifact name |
| _etl_operation_name | varchar | Brink operation name |
| _etl_ingested_at_utc | timestamptz | Ingestion timestamp |
| _etl_raw_s3_path | varchar | Raw S3 object path |
| _etl_raw_file_name | varchar | Raw source file name |
| _etl_parser_version | varchar | Parser version |
| _etl_schema_version | varchar | Schema version |
| _etl_contract_name | varchar | Brink contract name |
| _etl_contract_row_type | varchar | Parent/child row type discriminator |
| _etl_row_ordinal | bigint | Row ordinal within parsed payload |

## Table 2: `staging.brink_labor_breaks`

### Purpose
Raw break-level staging table. One row per Brink break, linked to the parent shift.

### Business columns
| Column name | Type | Notes |
|---|---|---|
| shift_id | varchar(36) | Parent Brink shift GUID |
| end_time | timestamptz | Brink `EndTime` |
| minutes | int | Brink `Minutes` |
| number | smallint | Brink `Number` |
| is_paid | boolean | Brink `IsPaid` |
| start_time | timestamptz | Brink `StartTime` |

### Technical columns
| Column name | Type | Notes |
|---|---|---|
| _etl_batch_id | varchar | Standard normalized metadata |
| _etl_company_id | varchar | Standard normalized metadata |
| _etl_tenant_id | varchar | Tenant identifier used by staging load |
| _etl_source_type | varchar | Standard normalized metadata |
| _etl_stream_name | varchar | Brink stream/artifact name |
| _etl_operation_name | varchar | Brink operation name |
| _etl_ingested_at_utc | timestamptz | Ingestion timestamp |
| _etl_raw_s3_path | varchar | Raw S3 object path |
| _etl_raw_file_name | varchar | Raw source file name |
| _etl_parser_version | varchar | Parser version |
| _etl_schema_version | varchar | Schema version |
| _etl_contract_name | varchar | Brink contract name |
| _etl_contract_row_type | varchar | Parent/child row type discriminator |
| _etl_row_ordinal | bigint | Row ordinal within parsed payload |

## Parquet / Row Model Structure

### Pattern
- Create shared staging/parquet row models, not separate parquet DTOs.
- Suggested models:
  - `BrinkLaborShiftsRow`
  - `BrinkLaborBreaksRow`
- Both inherit from `BrinkStagingRowBase`.
- Decimal properties should use current parquet precision attributes, consistent with existing Brink rows.

### Child collection handling
- `Breaks` must not stay nested in parquet.
- The normalizer emits:
  - one row in `brink_labor_shifts` per shift
  - one row in `brink_labor_breaks` per break
- `shift_id` is required in the child row to preserve the parent-child relationship.
- There is no Brink break identifier in the published contract, so no `break_id` column should be added for raw staging parity.

## Implementation Plan

### Module 1: Extract
- Add `Labor` to `BrinkServiceName` and `GetShifts` to `BrinkOperationName`.
- Add `BuildCaptureShifts(DateOnly businessDate)` to `BrinkOperationRequestBuilder` using:

```text
Path: Labor2.svc
SOAP Action: http://www.brinksoftware.com/webservices/labor/v2/ILaborWebService2/GetShifts
```

- Add SOAP envelope construction in `BrinkSoapEnvelopeBuilder` for the Labor v2 namespace.
- Extend `IBrinkRawClient` / `BrinkRawClient` with `CaptureShiftsAsync(...)`.
- Add `GetShifts` support to `BrinkPosEndpointCatalog` so the existing extract job includes the labor endpoint.
- Update `BrinkPosEndpointExecutor` to dispatch by endpoint definition instead of rejecting everything except `Sales2.GetOrders`.
- Keep `BrinkPosExtractOrchestrator` as the source of the endpoint list; no separate labor orchestrator should be introduced.
- Persist extract details as `StageDiagnosticsEnvelope<BrinkPosExtractBatchDetails>` so downstream stages can deserialize endpoint-level metadata safely.

### Module 2: Raw capture
- Persist raw XML via `IEtlS3Storage.WriteRawAsync(...)`.
- Use a stable S3 path convention for labor payloads.
- Keep raw content untouched.
- Preserve the same extract diagnostics shape already used by `BrinkPosEndpointExecutionResult`.

### Module 2A: Typed facade contract (optional but recommended for parity)
- Add `GetShiftsResponse` if the endpoint should be available through `IBrinkPosFacade`.
- Add a `LaborClient` mirroring the typed-client pattern already used for `Sales2Client`, `Settings2Client`, etc.
- Add `IBrinkPosFacade.GetShiftsAsync(...)` and implement it in `BrinkPosFacade`.
- Parse typed facade payloads with `BrinkXmlParser.GetOperationResultPayload(...)` and `BrinkXmlParser.GetCollectionElements(...)`.
- Register the typed client in `AddBrinkPosIntegration()`.

### Module 3: Normalize
- Add a dedicated staging profile, analogous to `Sales2GetOrdersStagingProfile`, likely `LaborGetShiftsStagingProfile`.
- Parse the SOAP body via `IBrinkXmlEnvelopeParser` and flatten the `Shifts -> Shift -> Breaks -> Break` structure.
- Emit one parent artifact for `brink_labor_shifts` and one child artifact for `brink_labor_breaks`.
- Apply standard Brink staging metadata through `BrinkStagingMetadataApplicator`.
- Map source fields 1:1 into flat `snake_case` business columns.
- Use `BrinkStagingArtifactDefinition` values aligned to the new contracts, e.g. `brink_labor_shifts` / `shift` and `brink_labor_breaks` / `shift_break`.
- Add the new profile registration in `AddBrinkPosEtlIntegration()`.
- Generalize `BrinkPosNormalizationContextFactory` so context values (`stream_name`, `operation_name`, `target_entity`, parser version) come from the extracted endpoint/contract being processed rather than hardcoded orders constants.

### Module 4: Parquet
- Add `BrinkLaborShiftsRow` and `BrinkLaborBreaksRow` models.
- Both models must inherit from `BrinkStagingRowBase`.
- Register `ParquetBatchWriter<T>` for both row types.
- Write parent and child artifacts separately.
- Compress parquet output and upload it for loading.
- Use the same decimal annotation pattern as existing Brink rows: `[ParquetDecimal(18, 4)]` for `declared_tips` and `pay_rate`.

### Module 5: Redshift staging load
- Add labor contracts to `BrinkPosNormalizationCatalog`.
- Add labor staging table metadata to `BrinkPosStagingTableCatalog`.
- Generalize `BrinkPosNormalizeLoadJobs` so it writes diagnostics for all configured Brink contracts required by the current raw extract, not only `GetSales2GetOrdersContracts()`.
- Generalize `BrinkPosLoadToStagingJobs` so it validates and loads the table set for the current Brink contract group, not only `GetSales2GetOrdersTables()`.
- Reuse existing `RedshiftStagingLoadExecutor`; no labor-specific load executor is needed.
- Parent delete predicate:

```sql
DELETE FROM staging.brink_labor_shifts
WHERE _etl_tenant_id = @tenant_id
  AND CAST(business_date AS DATE) = CAST(@business_date AS DATE);
```

- Child delete predicate:

```sql
DELETE FROM staging.brink_labor_breaks
WHERE _etl_tenant_id = @tenant_id
  AND shift_id IN (
      SELECT shift_id
      FROM staging.brink_labor_shifts
      WHERE _etl_tenant_id = @tenant_id
        AND CAST(business_date AS DATE) = CAST(@business_date AS DATE)
  );
```

- Ensure load is repeatable and idempotent.

### Module 6: SQL bootstrap
- Add Redshift bootstrap DDL scripts under `backend/Kds.Ai.Warehouse/Sql/Bootstrap/Tenant/Staging/BrinkPos/` for:
  - `staging.brink_labor_shifts`
  - `staging.brink_labor_breaks`
- Follow the current DDL pattern from `0605_create_staging_brink_sales2_get_orders.sql`:
  - business columns first
  - then all 14 `_etl_*` columns
  - `_etl_ingested_at_utc TIMESTAMP NOT NULL`
  - `_etl_row_ordinal BIGINT NOT NULL`

### Module 7: Tests
- Update `BrinkPosEndpointCatalogTests` to assert `GetShifts` is part of the v1 extract endpoint set.
- Update `BrinkPosEtlCatalogTests` because the default registration/catalog assumptions currently expect only Sales2 orders.
- Add raw transport tests for `BrinkRawClient` / `BrinkOperationRequestBuilder` to verify `Labor2.svc`, SOAP action, and envelope.
- Add staging profile tests for shifts and breaks, following `Sales2GetOrdersStagingProfileTests`.
- Add or update downstream live tests so the existing Brink pipeline can process the labor staging tables end-to-end.
- Add live facade tests if the typed facade surface is added:
  - `GetShifts_WhenConfigured_CallsRealBrinkEndpoint`
  - `CaptureShifts_WhenConfigured_StreamsRealBrinkEndpoint`
  These tests should write XML artifacts into `temp/` so the contract can be inspected directly.
- Add load/idempotency coverage for the new delete predicates.

## Contract Onboarding Checklist

Use this checklist for any new Brink endpoint/contract, not only `GetShifts`:

1. Add enum values in `BrinkServiceName` / `BrinkOperationName`.
2. Add SOAP envelope builder support.
3. Add raw request builder support.
4. Add raw capture method to `IBrinkRawClient` / `BrinkRawClient`.
5. Add endpoint to `BrinkPosEndpointCatalog`.
6. Add endpoint dispatch in `BrinkPosEndpointExecutor`.
7. Add operation metadata entry in `BrinkPosOperationCatalog`.
8. Add or update extract batch details contract if new metadata is required.
9. Add staging row model(s).
10. Add normalization profile and DI registration.
11. Add normalization contracts to `BrinkPosNormalizationCatalog`.
12. Add staging tables + idempotent delete predicates to `BrinkPosStagingTableCatalog`.
13. Add Redshift bootstrap SQL.
14. If typed facade access is needed, add client + facade interface + facade implementation + response model.
15. Add unit tests, ETL tests, and live tests.
16. Verify whether the endpoint should emit XML artifacts into `temp/` through facade/raw live tests.

## Notes
- This spec intentionally treats staging as raw landing data.
- No business enrichment, rollups, or normalization beyond flattening nested structures.
- This spec was adjusted to match current Brink implementation patterns: shared staging/parquet row models, `_etl_*` metadata, `snake_case` column names, and separate child artifacts for nested collections.
- This spec now also captures the repo-specific lessons from implementation: adding a new Brink contract is not only a row-model/DDL task. It usually touches transport, endpoint catalog, operation metadata, extract details serialization, normalization profile registration, staging catalogs, and optionally typed facade/live XML coverage.

## Code References
- `Kds.Ai.Integrations.BrinkPos.BrinkPosEndpointCatalog`
- `Kds.Ai.Integrations.BrinkPos.BrinkPosEndpointExecutor`
- `Kds.Ai.Integrations.BrinkPos.Models.BrinkPosEtlStageConstants`
- `Kds.Ai.Integrations.BrinkPos.BrinkPosNormalizationContextFactory`
- `Kds.Ai.Integrations.BrinkPos.BrinkPosNormalizationCatalog`
- `Kds.Ai.Integrations.BrinkPos.BrinkPosOperationCatalog`
- `Kds.Ai.Integrations.BrinkPos.BrinkPosLoadToStagingJobs`
- `Kds.Ai.Integrations.Etl.StageDiagnosticsEnvelope`
- `Kds.Ai.Integrations.Etl.StageDiagnosticsJson`
- `Kds.Ai.Warehouse.Models.Normalization.NormalizedRowBase`
- `Kds.Ai.Integrations.BrinkPos.Models.Staging.Common.BrinkStagingRowBase`
- `Kds.Ai.Integrations.BrinkPos.Services.Normalization.Staging.Common.BrinkStagingMetadataApplicator`
- `Kds.Ai.Integrations.BrinkPos.Services.Normalization.Staging.Profiles.Sales2GetOrdersStagingProfile`
- `Kds.Ai.Integrations.BrinkPos.Services.Facade.BrinkRawClient`
- `Kds.Ai.Integrations.BrinkPos.Services.Facade.Clients.LaborClient`
- `Kds.Ai.Integrations.BrinkPos.Models.Responses.GetShiftsResponse`
- `Kds.Ai.Integrations.BrinkPos.Services.Facade.BrinkOperationRequestBuilder`
- `Kds.Ai.Integrations.BrinkPos.Services.Facade.BrinkSoapEnvelopeBuilder`
- `Kds.Ai.Integrations.BrinkPos.Services.Facade.BrinkPosFacade`
- `Kds.Ai.Integrations.BrinkPos.BrinkPosStagingTableCatalog`