dv-bv-activity-schema

name	dv-bv-activity-schema
description	Design and deploy an Activity Schema 2.0 pipeline as a Business Vault satellite. Covers BV staging transformation view, SAT_BV_NH_{ENTITY}_STREAM DDL, stream-on-view, triggered tasks, and per-activity IM Dynamic Tables.
enabled	true

/dv-bv-activity-schema — Activity Schema as a Business Vault Satellite

This skill implements the Activity Schema 2.0 standard as a Business Vault (BV) pattern within DVOS. Event streams land in the Raw Vault, a BV staging transformation view maps raw events to named activities, and a triggered task populates the Activity Schema satellite — a standardised BV non-historised satellite with a fixed column set that any BI tool or Dynamic Table can query without knowing source-system internals.

---

What is Activity Schema in DVOS?

Activity Schema models an entity taking a sequence of activities over time. Each row represents one activity by one entity at one point in time. Examples:

• Customer 00034353 debited account on 2024-03-15 — amount $250
• Customer 00034353 issued statement on 2024-04-01

In DVOS terms this maps to two vault layers:

Layer	Object	Role
Raw Vault	SAT_NH_RV_{ENTITY}_{BADGE}	Stores the full raw event payload as dv_object VARIANT — nothing is lost
Business Vault	SAT_BV_NH_{ENTITY}_STREAM	Stores the interpreted activity: named activity, projected feature_json, revenue_impact

The BV satellite is the Activity Schema table. The RV satellite is the audit trail.

Key mapping (no duplicate columns needed):

Activity Schema column	DVOS equivalent
ts	dv_applied_timestamp
customer	business key column (e.g. customer_id)
activity	activity column in BV satellite
feature_json	feature_json VARIANT in BV satellite

---

Decision questions

Before generating any artefacts, confirm the following:

1. Entity — what is the entity? (customer, account, employee, vehicle…)
2. Parent hub — which hub anchors this stream? (HUB_{ENTITY})
3. Event codes → activity names — list each raw event code and the activity name it maps to:
   • e.g. event '1' → open_account, event '2' → debit_account
   • By convention, activity names are verb_noun from the entity's perspective
4. Authoritative attributes per activity — for each activity, which fields does it own?
   • Only attributes generated by that activity go into feature_json (AS spec rule)
   • e.g. debit_account owns: amount, currency, merchant_code — not address or account_type
5. Financial activities — which activities have a monetary impact? (revenue_impact populated for those; NULL for others)
6. Anonymous customer ID — is there a pre-resolution local identifier? (maps to the DVOS Same-As Link pattern)
7. Loading mode — Kappa Vault (stream-on-view, triggered task) or batch (scheduled task from RV satellite)?

---

Naming rules

Artefact	Pattern	Example
BV satellite	SAT_BV_NH_{ENTITY}_STREAM	SAT_BV_NH_CUSTOMER_STREAM
BV staging transformation view	stg_bv_{entity}_activity	stg_bv_customer_activity
Stream on BV staging view	str_bv_{entity}_activity_to_sat_bv_nh_{entity}_stream	str_bv_customer_activity_to_sat_bv_nh_customer_stream
Triggered task (RV → BV)	tsk_bv_{entity}_activity_to_sat_bv_nh_{entity}_stream	tsk_bv_customer_activity_to_sat_bv_nh_customer_stream
Per-activity IM Dynamic Table	dt_{entity}_stream_{activity}	dt_customer_stream_debit_account

No source badge on the BV satellite — it is derived from multiple sources, so no single badge applies. The SAT_BV_NH_ prefix and _STREAM suffix identify the pattern.

---

BV satellite DDL

-- Activity Schema BV satellite
CREATE OR REPLACE TRANSIENT TABLE <schema>.SAT_BV_NH_<ENTITY>_STREAM
(
    -- DVOS metadata columns
    dv_tenant_id             VARCHAR(50),
    dv_collisioncode         VARCHAR(50),
    dv_hashkey_hub_<entity>  BINARY(20)    NOT NULL,
    dv_load_timestamp        TIMESTAMP_NTZ NOT NULL,
    dv_applied_timestamp     TIMESTAMP_NTZ NOT NULL,  -- maps to AS 'ts'
    dv_recordsource          VARCHAR(255)  NOT NULL,
    dv_task_id               VARCHAR(255),
    dv_jira_id               VARCHAR(255),
    dv_user_id               VARCHAR(255),
    dv_sid                   NUMBER        IDENTITY START 0 INCREMENT 1 ORDER,  -- SNOPIT locator

    -- Business key (maps to AS 'customer')
    <bk_column>              VARCHAR(50),

    -- Activity Schema columns
    activity_id              VARCHAR(50)   NOT NULL,  -- unique event identifier
    activity                 VARCHAR(50)   NOT NULL,  -- verb_noun from entity perspective
    anonymous_customer_id    VARCHAR(50)   NULL,      -- pre-resolution local id (optional)
    feature_json             VARIANT       NOT NULL,  -- authoritative attributes only
    revenue_impact           NUMBER(18,2)  NULL,      -- NULL for non-financial activities
    link                     VARCHAR(255)  NULL,      -- URL to source record (optional)

    CONSTRAINT pk_sat_bv_nh_<entity>_stream
        PRIMARY KEY (dv_hashkey_hub_<entity>, dv_load_timestamp) NOT ENFORCED
)
DATA_RETENTION_TIME_IN_DAYS = 7;

-- Ghost record (required for SNOPIT/PIT equi-joins)
INSERT INTO <schema>.SAT_BV_NH_<ENTITY>_STREAM
    (dv_tenant_id, dv_hashkey_hub_<entity>, dv_load_timestamp, dv_applied_timestamp,
     dv_recordsource, dv_task_id, dv_jira_id, activity_id, activity, feature_json,
     revenue_impact, link)
SELECT
    NULL,
    TO_BINARY(REPEAT(0, 20)),
    TO_TIMESTAMP('1900-01-01 00:00:00'),
    TO_TIMESTAMP('1900-01-01 00:00:00'),
    'GHOST', 'GHOST', 'GHOST', 'GHOST', 'GHOST',
    PARSE_JSON('{}'), NULL, 'GHOST';

VC_ and VH_ helper views

-- Current state view
CREATE OR REPLACE VIEW <schema>.VC_SAT_BV_NH_<ENTITY>_STREAM AS
SELECT *,
    COALESCE(LEAD(dv_applied_timestamp) OVER (PARTITION BY dv_hashkey_hub_<entity>
             ORDER BY dv_applied_timestamp, dv_load_timestamp), CAST('9999-12-31' AS DATE)) AS dv_applied_timestamp_end,
    CASE WHEN LEAD(dv_applied_timestamp) OVER (PARTITION BY dv_hashkey_hub_<entity>
         ORDER BY dv_applied_timestamp, dv_load_timestamp) IS NULL THEN 1 ELSE 0 END        AS dv_currentflag
FROM <schema>.SAT_BV_NH_<ENTITY>_STREAM
QUALIFY ROW_NUMBER() OVER (PARTITION BY dv_hashkey_hub_<entity>
                           ORDER BY dv_applied_timestamp DESC, dv_load_timestamp DESC) = 1;

-- History view
CREATE OR REPLACE VIEW <schema>.VH_SAT_BV_NH_<ENTITY>_STREAM AS
SELECT *,
    COALESCE(LEAD(dv_applied_timestamp) OVER (PARTITION BY dv_hashkey_hub_<entity>
             ORDER BY dv_applied_timestamp, dv_load_timestamp), CAST('9999-12-31' AS DATE)) AS dv_applied_timestamp_end,
    CASE WHEN LEAD(dv_applied_timestamp) OVER (PARTITION BY dv_hashkey_hub_<entity>
         ORDER BY dv_applied_timestamp, dv_load_timestamp) IS NULL THEN 1 ELSE 0 END        AS dv_currentflag
FROM <schema>.SAT_BV_NH_<ENTITY>_STREAM;

---

BV staging transformation view

This view reads from the Raw Vault satellite and maps raw events to Activity Schema format. It is the modeling step — event codes become business concepts here.

CREATE OR REPLACE VIEW <schema>.stg_bv_<entity>_activity AS
SELECT
    -- DVOS metadata pass-through
    dv_tenant_id,
    dv_hashkey_hub_<entity>,
    dv_load_timestamp,
    dv_applied_timestamp,
    dv_recordsource,
    dv_task_id,
    dv_jira_id,
    dv_user_id,
    <bk_column>,

    -- Activity Schema: unique event identifier
    PARSE_JSON(dv_object):'<event_id_field>'::TEXT                      AS activity_id,

    -- Activity Schema: map event codes to named activities
    CASE
        WHEN PARSE_JSON(dv_object):'<event_field>'::TEXT = '<code_1>'  THEN '<activity_1>'
        WHEN PARSE_JSON(dv_object):'<event_field>'::TEXT = '<code_2>'  THEN '<activity_2>'
        WHEN PARSE_JSON(dv_object):'<event_field>'::TEXT = '<code_3>'  THEN '<activity_3>'
    END                                                                 AS activity,

    -- Anonymous ID (NULL if not applicable)
    NULL::VARCHAR(50)                                                   AS anonymous_customer_id,

    -- feature_json: ONLY attributes authoritative to this activity
    -- Use OBJECT_CONSTRUCT — never pass dv_object directly
    CASE
        WHEN PARSE_JSON(dv_object):'<event_field>'::TEXT = '<code_1>'
            THEN OBJECT_CONSTRUCT(
                '<attr_1a>', PARSE_JSON(dv_object):'<path_1a>',
                '<attr_1b>', PARSE_JSON(dv_object):'<path_1b>'
            )
        WHEN PARSE_JSON(dv_object):'<event_field>'::TEXT = '<code_2>'
            THEN OBJECT_CONSTRUCT(
                '<attr_2a>', PARSE_JSON(dv_object):'<path_2a>'
            )
        ELSE OBJECT_CONSTRUCT()
    END                                                                 AS feature_json,

    -- revenue_impact: NULL for non-financial activities
    CASE
        WHEN PARSE_JSON(dv_object):'<event_field>'::TEXT IN ('<financial_code_1>', '<financial_code_2>')
            THEN PARSE_JSON(dv_object):'<amount_path>'::FLOAT
        ELSE NULL
    END                                                                 AS revenue_impact,

    -- link: URL to source record if available
    NULL::VARCHAR(255)                                                  AS link

FROM <schema>.SAT_NH_RV_<ENTITY>_<BADGE>
WHERE PARSE_JSON(dv_object):'<event_field>'::TEXT
    IN ('<code_1>', '<code_2>', '<code_3>');

Rule: use OBJECT_CONSTRUCT — never pass dv_object or dv_object:"details" directly as feature_json. Explicit projection enforces the AS spec's authoritative-attributes-only rule and makes the contract visible in the view definition.

---

Stream on BV staging view

-- Kappa Vault: stream on the transformation view
CREATE OR REPLACE STREAM <schema>.str_bv_<entity>_activity_to_sat_bv_nh_<entity>_stream
ON VIEW <schema>.stg_bv_<entity>_activity
APPEND_ONLY    = TRUE
SHOW_INITIAL_ROWS = TRUE;

---

Triggered task — RV to BV

CREATE OR REPLACE TASK <schema>.tsk_bv_<entity>_activity_to_sat_bv_nh_<entity>_stream
    WAREHOUSE = <wh>
    WHEN SYSTEM$STREAM_HAS_DATA('<schema>.str_bv_<entity>_activity_to_sat_bv_nh_<entity>_stream')
AS
INSERT INTO <vault_schema>.SAT_BV_NH_<ENTITY>_STREAM
    (dv_tenant_id, dv_hashkey_hub_<entity>, dv_load_timestamp, dv_applied_timestamp,
     dv_recordsource, dv_task_id, dv_jira_id, dv_user_id, <bk_column>,
     activity_id, activity, anonymous_customer_id,
     feature_json, revenue_impact, link)
SELECT
    dv_tenant_id,
    dv_hashkey_hub_<entity>,
    dv_load_timestamp,
    dv_applied_timestamp,
    dv_recordsource,
    dv_task_id,
    dv_jira_id,
    dv_user_id,
    <bk_column>,
    activity_id,
    activity,
    anonymous_customer_id,
    feature_json,
    revenue_impact,
    link
FROM <schema>.str_bv_<entity>_activity_to_sat_bv_nh_<entity>_stream;
-- No NOT EXISTS check — stream is append-only, every row is new

ALTER TASK <schema>.tsk_bv_<entity>_activity_to_sat_bv_nh_<entity>_stream RESUME;

---

Per-activity IM Dynamic Table

One Dynamic Table per activity. TARGET_LAG = DOWNSTREAM — refreshed when its upstream stream refreshes.

CREATE OR REPLACE DYNAMIC TABLE <im_schema>.dt_<entity>_stream_<activity>
    TARGET_LAG = DOWNSTREAM
    WAREHOUSE  = <wh>
AS
WITH activity_agg AS (
    SELECT <bk_column>, COUNT(*) AS activity_count
    FROM <vault_schema>.SAT_BV_NH_<ENTITY>_STREAM
    WHERE activity = '<activity>'
    GROUP BY 1
)
SELECT
    bv.*,
    ROW_NUMBER()  OVER (PARTITION BY bv.<bk_column>, bv.activity
                        ORDER BY bv.dv_applied_timestamp)                   AS activity_occurrence,
    COALESCE(
        LAG(bv.dv_applied_timestamp) OVER (PARTITION BY bv.<bk_column>, bv.activity
                                     ORDER BY bv.dv_applied_timestamp),
        '1900-01-01'::DATE)                                           AS activity_previous_at,
    COALESCE(
        LEAD(bv.dv_applied_timestamp) OVER (PARTITION BY bv.<bk_column>, bv.activity
                                      ORDER BY bv.dv_applied_timestamp),
        '9999-12-31'::DATE)                                           AS activity_repeated_at,
    agg.activity_count
FROM <vault_schema>.SAT_BV_NH_<ENTITY>_STREAM bv
INNER JOIN activity_agg agg
    ON bv.<bk_column> = agg.<bk_column>
WHERE bv.activity = '<activity>';

For relationship queries (first before, last before, aggregate before/after, ASOF dim enrichment) — use /dv-mart which generates the enriched relationship Dynamic Table from per-activity DTs.

---

Batch loading variant (non-Kappa)

If not using Kappa Vault, replace the stream + triggered task with a scheduled task reading directly from the RV satellite:

CREATE OR REPLACE TASK <schema>.tsk_bv_<entity>_activity_daily
    WAREHOUSE = <wh>
    SCHEDULE  = 'USING CRON 0 6 * * * Australia/Sydney'
AS
INSERT INTO <vault_schema>.SAT_BV_NH_<ENTITY>_STREAM
    (dv_tenant_id, dv_hashkey_hub_<entity>, dv_load_timestamp, dv_applied_timestamp,
     dv_recordsource, dv_task_id, dv_jira_id, dv_user_id, <bk_column>,
     activity_id, activity, anonymous_customer_id,
     feature_json, revenue_impact, link)
SELECT ...
FROM <schema>.stg_bv_<entity>_activity
WHERE dv_load_timestamp > (SELECT COALESCE(MAX(dv_load_timestamp), '1900-01-01')
                     FROM <vault_schema>.SAT_BV_NH_<ENTITY>_STREAM);

---

Flow diagram

landed.{source}           VARIANT payload (Kafka / Snowpipe / COPY INTO)
        |
staged.stg_{source}       DVOS metadata tags — hashkey, bkcc, dv_applied_timestamp
        |
        +-- stream --> HUB_{ENTITY} (triggered task)
        |
        +-- stream --> SAT_NH_RV_{ENTITY}_{BADGE} (triggered task, dv_object VARIANT)
                          |
                datavault.stg_bv_{entity}_activity   BV transformation view
                   event codes → activity names
                   OBJECT_CONSTRUCT for feature_json
                          |
                          +-- stream --> SAT_BV_NH_{ENTITY}_STREAM (triggered task)
                                             |
                information_marts.dt_{entity}_stream_{activity}   per-activity DT
                                             |
                information_marts.dt_{entity}_stream_enriched     relationship DT (/dv-mart)
                                             |
                information_marts.dt_{entity}_stream_enriched_with_{dim}  ASOF dim join (/dv-mart)

---

Subagent files

• SQL Generator: agents/sql-generator.md — Activity Schema DDL and view templates
• Naming Advisor: agents/naming-advisor.md — enforces SAT_BV_NH_{ENTITY}_STREAM pattern