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

Source Contract

Shift data

Break data

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
2. Add service/operation enum values.
3. Add SOAP envelope builder support.
4. Add raw request builder support.
5. Add raw capture method to IBrinkRawClient / BrinkRawClient.
6. Add endpoint definition to BrinkPosEndpointCatalog.
7. Update BrinkPosEndpointExecutor to dispatch the new endpoint.

1. Operation metadata
2. Add the contract group to BrinkPosOperationCatalog.
3. Define:
4. stream_name
5. operation_name
6. target_contract_group
7. parser_version
8. request metadata
9. contract list
10. staging table list

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

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

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

1. Optional typed facade support
2. If the endpoint should be available through IBrinkPosFacade, add:
3. request model if needed
4. typed response model
5. service client (for example LaborClient)
6. facade interface method
7. facade implementation
8. DI registration

1. Validation coverage
2. Add unit transport coverage.
3. Add staging-profile coverage.
4. Add ETL catalog / executor coverage.
5. Add live ETL coverage when applicable.
6. 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

Technical columns

Table 2: staging.brink_labor_breaks

Purpose

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

Business columns

Technical columns

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:

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:

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:

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