# Bulk Asset Ingester Scheduler (bulk-assets-ingester-sfn Scheduler)

## Brief
This is a convenience script to abstract and document the SQL and logic used to schedule and process large amounts of asset ingestion data.

This script will take a table containing pre-formed release asset information (of arbitrary size) and split it into sub-tables of requested size (default 400). The tables will then be used to generate JSON documents which trigger and feed `{env}-bulk-assets-ingester-sfn`.

## Reason
This is necessary, as there is an execution limit for State Machines on AWS of 25,000 events. By splitting a large table of many thousands of rows into smaller tables, and putting those table names in an extra-memory queue (on Snowflake in this implementation), we  can exert some control over the max number of executions for any state machine, while keeping ingestion operations tightly correlated from source through final result. (A source table -> an execution table -> an SFN execution)

## .env.shadow
These are the env vars expected by the script.
```
# INT-887-Bulk_Asset_Scheduler - Env
ENVIRONMENT=dev

# AWS ACCESS
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=

# AWS SFN
SFN_NAME=

# Snowflake
SNOWFLAKE_USER=
SNOWFLAKE_PASSWORD=
SNOWFLAKE_CATALOG=
SNOWFLAKE_SCHEMA=
SNOWFLAKE_KEY_PASSPHRASE=

# Script Vars
# Max num of assets to put in each split sub-table - Max 500, or SFN fails
ASSETS_PER_TABLE=
# Bucket to plac JSON trigger file
JSON_DROP_BUCKET=
# The folder to put the JSON trigger file on the above bucket
JSON_DROP_KEY_PREFIX=
# How many SFN's to allow running at once.
MAX_CONCURRENT_SFN=

# Create a table with one of each asset row, and only use that.
DEBUG_WAKEUP=False
```

## Data Store Mechanics
In practice this script is fed from a single table, stored on Snowflake, in a per-client schema under the INTEGRATION catalog. This could of course be ported to any Snowflake cat/schema combo.

The table `{env}_BULK_ASSET_TABLE_QUEUE` is a simple minimal column table that stores Source Table names. ("Source Tables" are the full sized tables with all rows and are defined with examples in an appendix below.)

Table Queue DDL:
```

-- Create PROD source table queue - SOURCE TABLE NAMES GO HERE TO BE SPLIT
CREATE TABLE IF NOT EXISTS QA_BULK_ASSET_TABLE_QUEUE (
	ID INTEGER AUTOINCREMENT,
	SOURCE_TABLE VARCHAR(255) NOT NULL,
	CONSTRAINT QA_BULK_ASSET_INGESTION_QUEUE_PK PRIMARY KEY (ID)
);
COMMENT ON COLUMN QA_BULK_ASSET_TABLE_QUEUE.ID IS 'primary key';
COMMENT ON COLUMN QA_BULK_ASSET_TABLE_QUEUE.SOURCE_TABLE IS 'The table which contains the asset information.';
```

The script logic reads this table, proceeds to calculate how many assets are presented per release, and splits the table into sub-tables such that the maximum count of rows per table is defined in the envirnoment var `ASSETS_PER_TABLE`. This value is an upper bound, e.g. the logic ensures that releases are not split between tables, even if a table's row count doesn't reach the `ASSETS_PER_TABLE` value. 

These sub-tables names are generated from the source table name, appended with `_xxx` where `xxx` are digits beginning at `001`. The DDL of each sub-table matches the DDL of the source table.

The sub-table names, and some correlated metadata is stored in `{env}_BULK_ASSET_SFN_QUEUE` 

```
-- Example INTEGRATION.ABOVEBOARD.PROD_BULK_ASSET_SFN_QUEUE definition

create or replace TABLE INTEGRATION.ABOVEBOARD.QA_BULK_ASSET_SFN_QUEUE (
	ID NUMBER(38,0) NOT NULL autoincrement,
	SFN_NAME VARCHAR(255) NOT NULL,
	SOURCE_TABLE VARCHAR(255) NOT NULL,
	TRIGGER_JSON_KEY VARCHAR(255),
	EXECUTION_NAME VARCHAR(255),
	STATUS VARCHAR(255),
	QUEUE_TIME TIMESTAMP_LTZ(9) NOT NULL,
	START_TIME TIMESTAMP_LTZ(9),
	END_TIME TIMESTAMP_LTZ(9),
	constraint QA_BULK_ASSET_INGESTION_QUEUE_PK primary key (ID)
);
```

## Runtime
At runtime, the script will go through the following steps:

1. Create queue tables
1. Look for new source tables, make sub-tables, and queue them
1. Delete table from source table queue
1. Find and log missing executions
1. Update in-progress SFN execution statuses in the queue
1. Execute each table in the SFN queue until MAX_CONCURRENT_SFN jobs are running.
1. (Attempt to) Log the execution name.

## Execution
This job is designed as a finite state system to be run on a chron schedule. Each execution will read the state of the system (Current AWS SFN execution count, new / outstanding items in queue, etc.) and execute whatever steps from the list above are required in the defined order.

It can be run ad-hoc as needed.

The inciting event is the presence of a source table name in `{env}_BULK_ASSET_TABLE_QUEUE`. All states stem from this event.

## Apendices

### Source Table Structure
Here is an example definition for an actual "Source Table" used with this script.
```
-- Example INTEGRATION.ABOVEBOARD.ASSET_ALL_BATCH12 definition

create or replace TABLE INTEGRATION.ABOVEBOARD.ASSET_ALL_BATCH12 (
	FOREIGN_REL_ID VARCHAR(16777216),  -- unique key from original data
	VOLUME NUMBER(38,0),
	TRACK NUMBER(38,0),
	FILE_NAME VARCHAR(16777216),
	PATH VARCHAR(16777216),  -- S3 path
	UPLOAD_DATE DATE,
	ORCH_UPC NUMBER(20,0),
	ORCH_RELEASE_ID NUMBER(38,0),
	ORCH_PROJECT_CODE VARCHAR(16777216),
	VENDOR_ID NUMBER(38,0),
	SESSION_ID VARCHAR(1),
	ASSET_TYPE VARCHAR(5),  -- like enum (cover, audio)
	FILE_SIZE NUMBER(38,0),
	S3_BUCKET VARCHAR(19)
);
```
#### Sample data:
|FOREIGN_REL_ID|VOLUME|TRACK|FILE_NAME|PATH|UPLOAD_DATE|ORCH_UPC|ORCH_RELEASE_ID|ORCH_PROJECT_CODE|VENDOR_ID|SESSION_ID|ASSET_TYPE|FILE_SIZE|S3_BUCKET|
|--------------|------|-----|---------|----|-----------|--------|---------------|-----------------|---------|----------|----------|---------|---------|
|5060705290026|1|1|5060705290026_1_01_wav.wav|AboveBoard/test/6894703125-5060705290026|2023-02-10|5060705290026|4377155|6894703121|74344||audio|83119724|prod-orcd-fuga-drop|
|5060705290026|1|2|5060705290026_1_02_wav.wav|AboveBoard/test/6894703125-5060705290026|2023-02-10|5060705290026|4377155|6894703121|74344||audio|94770800|prod-orcd-fuga-drop|