Editing Indici data

==Background==
[[Indici]] data is received automatically every day from Valentia. Ultimately, it is envisaged that this will form an important part of a broader K'aute data platform. But, for now, the focus is on picking up the data and bringing it to a manageable state so it can be used for analysis and contract reporting.

==Data transfer==

Summary of the indici data transfer process is shown in the graph below:

[[File:Indici_load_updated.drawio.png|800x800px]]


Valentia sends daily delta files (Parquet format) to <code>arn:aws:s3:::kpa-valentia</code>. The IAM user <code>valentia</code> uses policy <code>valentia-import</code> for:
* <code>s3:ListBucket</code>
* <code>s3:PutObject</code>
* <code>s3:GetObject</code>

'''Trigger mechanism:'''
An Amazon EventBridge rule detects file uploads and starts the Step Functions workflow. Lifecycle policies:
# Move files to Intelligent tiering after a few days
# Move to Archive format after longer periods

==Partitioning incoming data==

When EventBridge detects a new file:
# Triggers Step Functions workflow
# Workflow calls <code>copyIndiciFiles</code> lambda
# File moved to date-partitioned location in <code>kpa-indici-partitioned</code>

Files are partitioned using their embedded timestamp. All files are processed regardless of content.<ref>Empty files are handled in later steps</ref>

==Data processing workflow==

The Step Functions workflow coordinates:

=== Loading data ===
<code>indiciLoadAppointmentMedications</code> lambda:
# Reads Parquet → pandas DataFrame
# Processes datetimes (NaT → 'None')
# Identifies dataset from filename
# Pulls transformation rules from DynamoDB
# Adds metadata (filename + timestamp)
# Validates row count > 0
# Executes SQL to load into <code>indici_staging</code>
# Logs to <code>auditlog</code>

''DB role:'' <code>lambda_writer</code><br>
''Requires:'' [[Psycopg2 layer]]

=== Deduplication ===
<code>rptDeduplication</code> lambda:
# Performs UPSERTs to <code>rpt</code> tables
# Handles conflicts using primary keys
# Logs update/insert counts to <code>auditlog</code>

==Error handling==
Workflow handles:
* Empty files → Logs to CloudWatch and skips
* Missing SQL definitions → Triggers static data loading
* Other errors → Fails workflow and logs details

==Audit logging==
All outcomes logged through <code>stateMachine_LogHandler</code> to CloudWatch:
* <code>SuccessLogs</code>: Processed files
* <code>EmptyFilesLogs</code>: Skipped empty files
* <code>ErrorLogs</code>: Processing failures

==Infrastructure==
'''Obsolete:'''
* <s><code>pg_cron</code> jobs</s>
* <s>Scheduled database functions</s>

'''Current:'''
* EventBridge rule
* Step Functions state machine
* CloudWatch log groups
* Processing Lambdas

==Version control==
{{See also|Source control}}
All related code for this work should be checked into gitlab repo [https://gitlab.com/kaute-pasifika/malamalama malamalama].

==TODO==
* Update/Insert counts should be logged in <code>indici_staging</code>.<code>auditlog</code>. - Not sure if this has been implemented?

==References==

[[Category:Indici]]