Correct account lookup bug

This reverts commit 6d2150d1b2.

.dockerignore

@@ -28,6 +28,7 @@
 **/values.dev.yaml
 /bin
 /target
+!/target/**/monzo-ingestion
 /.idea
 LICENSE
 README.md

.env

@@ -1 +1 @@
-DATABASE_URL=postgres://postgres@localhost/logos_neu
+DATABASE_URL=postgres://postgres@localhost/monzo_development

.github/workflows/build.yml

@@ -1,33 +1,102 @@
-name: Build and Publish Docker Container
+name: Build and Publish
 
 on:
   push:
-    branches:
+    branches: [ main ]
-      - main
+
+env:
+  CARGO_TERM_COLOR: always
+  RUST_BINARY_NAME: monzo-ingestion
 
 jobs:
   build:
+    name: Build and Test
     runs-on: ubuntu-latest
-    container:
+    container: catthehacker/ubuntu:act-latest
-      image: catthehacker/ubuntu:act-latest
 
     steps:
-      - name: Checkout code
+      - uses: actions/checkout@v3
-        uses: actions/checkout@v2
+
+      - name: Install Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          profile: minimal
+          override: true
+          components: rustfmt, clippy
+
+      - name: Add ARM64 target
+        run: rustup target add aarch64-unknown-linux-musl
+
+      - name: Install ARM64 toolchain
+        run: |
+          apt-get update
+          apt-get install -y gcc-aarch64-linux-gnu build-essential musl-tools
+
+      - name: Cache dependencies
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cargo
+            target/
+          key: "${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}"
+          restore-keys: |
+            ${{ runner.os }}-cargo-            
+
+      - name: Build (x86_64)
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release --all-features
+
+      - name: Build (ARM64)
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release --all-features --target aarch64-unknown-linux-musl
+        env:
+          CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER: aarch64-linux-gnu-gcc
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v2
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v1
+        uses: docker/setup-buildx-action@v2
 
-      - name: Login to Docker
+      - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v2
         with:
           registry: git.joshuacoles.me
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
 
-      - name: Build and Push Docker image
+      - name: Upload artifacts
-        uses: docker/build-push-action@master
+        uses: actions/upload-artifact@v3
+        with:
+          name: binaries
+          path: |
+            target/release/${{ env.RUST_BINARY_NAME }}
+            target/aarch64-unknown-linux-musl/release/${{ env.RUST_BINARY_NAME }}            
+
+#      # The target directory is kept in the .dockerignore file, so allow these to be copied in we need to move them to a
+#      # new directory.
+#      - run: mv target docker-binaries
+
+      - name: Build and push multi-arch Docker image
+        uses: docker/build-push-action@v4
         with:
           context: .
+          file: ./Dockerfile
+          platforms: linux/amd64,linux/arm64
           push: true
-          tags: git.joshuacoles.me/personal/monzo-ingestion:latest
+          tags: git.joshuacoles.me/${{ github.repository }}:${{ github.sha }},git.joshuacoles.me/${{ github.repository }}:latest
+          build-args: |
+            BINARY_NAME=${{ env.RUST_BINARY_NAME }}
+
+      - uses: robiningelbrecht/ntfy-action@v1.0.0
+        name: Notify via ntfy.sh
+        if: always()
+        with:
+          url: ${{ secrets.NTFY_URL }}
+          topic: ${{ secrets.NTFY_TOPIC }}
+          job_status: ${{ job.status }}

Cargo.toml

@@ -7,29 +7,31 @@ edition = "2021"
 entity = { path = "entity" }
 migration = { path = "migration" }
 
-axum = { version = "0.7.5", features = ["multipart"] }
+axum = { version = "0.8.8", features = ["multipart"] }
-tokio = { version = "1.37.0", features = ["full"] }
+tokio = { version = "1.48.0", features = ["full"] }
-sea-orm = { version = "1.0.0-rc.4", features = [
+sea-orm = { version = "1.1.0", features = [
     "sqlx-postgres",
     "runtime-tokio-rustls",
     "macros"
 ] }
 
-serde = { version = "1.0.203", features = ["derive"] }
+serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0.117"
+serde_json = "1.0"
 tracing-subscriber = "0.3.18"
 tracing = "0.1.40"
-anyhow = "1.0.86"
+anyhow = { version = "1.0", features = ["backtrace"] }
-thiserror = "1.0.61"
+thiserror = "2.0"
-http = "1.1.0"
+http = "1.1"
-chrono = { version = "0.4.38", features = ["serde"] }
+chrono = { version = "0.4", features = ["serde"] }
-num-traits = "0.2.19"
+num-traits = "0.2"
 csv = "1.3.0"
-clap = "4.5.4"
+clap = "4.5"
-testcontainers = "0.17.0"
+testcontainers = "0.26"
-testcontainers-modules = { version = "0.5.0", features = ["postgres"] }
+testcontainers-modules = { version = "0.14", features = ["postgres"] }
-sqlx = { version = "0.7.4", features = ["postgres"] }
+sqlx = { version = "0.8", features = ["postgres"] }
-tower-http = { version = "0.5.2", features = ["trace"] }
+tower-http = { version = "0.6", features = ["trace"] }
+bytes = "1.7"
+once_cell = "1.19"
 
 [workspace]
 members = [".", "migration", "entity"]

Dockerfile

@@ -1,78 +1,14 @@
-# syntax=docker/dockerfile:1
+FROM busybox AS platform_determiner
+ARG TARGETPLATFORM
+ARG BINARY_NAME
 
-# Comments are provided throughout this file to help you get started.
+COPY /target /target
-# If you need more help, visit the Dockerfile reference guide at
+RUN case "$TARGETPLATFORM" in \
-# https://docs.docker.com/engine/reference/builder/
+      "linux/amd64") BINARY_PATH="/target/release/${BINARY_NAME}" ;; \
+      "linux/arm64") BINARY_PATH="/target/aarch64-unknown-linux-musl/release/${BINARY_NAME}" ;; \
+      *) exit 1 ;; \
+    esac && mv "$BINARY_PATH" "/usr/bin/monzo-ingestion" && chmod +x "/usr/bin/monzo-ingestion"
 
-################################################################################
+FROM --platform=$TARGETPLATFORM debian:bookworm-slim
-# Create a stage for building the application.
+COPY --from=platform_determiner /usr/bin/monzo-ingestion /usr/local/bin/monzo-ingestion
-
+ENTRYPOINT ["/usr/local/bin/monzo-ingestion"]
-ARG RUST_VERSION=1.76.0
-ARG APP_NAME=monzo-ingestion
-FROM rust:${RUST_VERSION}-slim-bullseye AS build
-ARG APP_NAME
-WORKDIR /app
-
-# Build the application.
-# Leverage a cache mount to /usr/local/cargo/registry/
-# for downloaded dependencies and a cache mount to /app/target/ for 
-# compiled dependencies which will speed up subsequent builds.
-# Leverage a bind mount to the src directory to avoid having to copy the
-# source code into the container. Once built, copy the executable to an
-# output directory before the cache mounted /app/target is unmounted.
-RUN --mount=type=bind,source=src,target=src \
-    --mount=type=bind,source=entity,target=entity \
-    --mount=type=bind,source=migration,target=migration \
-    --mount=type=bind,source=Cargo.toml,target=Cargo.toml \
-    --mount=type=bind,source=Cargo.lock,target=Cargo.lock \
-    --mount=type=cache,target=/app/target/ \
-    --mount=type=cache,target=/usr/local/cargo/registry/ \
-    <<EOF
-set -e
-cargo build --locked --release
-cp ./target/release/$APP_NAME /bin/server
-EOF
-
-################################################################################
-# Create a new stage for running the application that contains the minimal
-# runtime dependencies for the application. This often uses a different base
-# image from the build stage where the necessary files are copied from the build
-# stage.
-#
-# The example below uses the debian bullseye image as the foundation for running the app.
-# By specifying the "bullseye-slim" tag, it will also use whatever happens to be the
-# most recent version of that tag when you build your Dockerfile. If
-# reproducability is important, consider using a digest
-# (e.g., debian@sha256:ac707220fbd7b67fc19b112cee8170b41a9e97f703f588b2cdbbcdcecdd8af57).
-FROM debian:bullseye-slim AS final
-
-RUN set -ex; \
-    apt-get update && \
-    apt-get -y install --no-install-recommends \
-        ca-certificates curl && \
-    rm -rf /var/lib/apt/lists/*
-
-# Create a non-privileged user that the app will run under.
-# See https://docs.docker.com/develop/develop-images/dockerfile_best-practices/#user
-ARG UID=10001
-RUN adduser \
-    --disabled-password \
-    --gecos "" \
-    --home "/nonexistent" \
-    --shell "/sbin/nologin" \
-    --no-create-home \
-    --uid "${UID}" \
-    appuser
-USER appuser
-
-# Copy the executable from the "build" stage.
-COPY --from=build /bin/server /bin/
-
-# Expose the port that the application listens on.
-EXPOSE 3000
-
-HEALTHCHECK --interval=5s --timeout=3s --retries=3 \
-    CMD curl -f http://localhost:3000/health || exit 1
-
-# What the container should run when it is started.
-CMD ["/bin/server"]

entity/Cargo.toml

@@ -9,5 +9,5 @@ name = "entity"
 path = "src/lib.rs"
 
 [dependencies]
-sea-orm = { version = "1.0.0-rc.4" }
+sea-orm = { version = "1.1.0" }
 serde = { version = "1.0.203", features = ["derive"] }

entity/src/account.rs

@@ -0,0 +1,26 @@
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "account")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub name: String,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(has_many = "super::transaction::Entity")]
+    Transaction,
+}
+
+impl Related<super::transaction::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Transaction.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}

entity/src/expenditure.rs

@@ -1,4 +1,4 @@
-//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
 
 use sea_orm::entity::prelude::*;
 use serde::{Deserialize, Serialize};
@@ -14,6 +14,21 @@ pub struct Model {
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
-pub enum Relation {}
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::transaction::Entity",
+        from = "Column::TransactionId",
+        to = "super::transaction::Column::Id",
+        on_update = "NoAction",
+        on_delete = "Cascade"
+    )]
+    Transaction,
+}
+
+impl Related<super::transaction::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Transaction.def()
+    }
+}
 
 impl ActiveModelBehavior for ActiveModel {}

entity/src/lib.rs

@@ -1,6 +1,7 @@
-//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
 
 pub mod prelude;
 
+pub mod account;
 pub mod expenditure;
 pub mod transaction;

entity/src/prelude.rs

@@ -1,4 +1,5 @@
-//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
 
+pub use super::account::Entity as Account;
 pub use super::expenditure::Entity as Expenditure;
 pub use super::transaction::Entity as Transaction;

entity/src/transaction.rs

@@ -1,4 +1,4 @@
-//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.0
 
 use sea_orm::entity::prelude::*;
 use serde::{Deserialize, Serialize};
@@ -11,14 +11,40 @@ pub struct Model {
     pub transaction_type: String,
     pub total_amount: Decimal,
     pub timestamp: DateTime,
-    pub title: String,
+    pub title: Option<String>,
     pub emoji: Option<String>,
     pub notes: Option<String>,
     pub receipt: Option<String>,
     pub description: Option<String>,
+    #[sea_orm(unique)]
+    pub identity_hash: Option<i64>,
+    pub account_id: Option<i32>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
-pub enum Relation {}
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::account::Entity",
+        from = "Column::AccountId",
+        to = "super::account::Column::Id",
+        on_update = "NoAction",
+        on_delete = "Cascade"
+    )]
+    Account,
+    #[sea_orm(has_many = "super::expenditure::Entity")]
+    Expenditure,
+}
+
+impl Related<super::account::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Account.def()
+    }
+}
+
+impl Related<super::expenditure::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Expenditure.def()
+    }
+}
 
 impl ActiveModelBehavior for ActiveModel {}

fixtures/transactions.csv

@@ -0,0 +1,3 @@
+Transaction ID,Date,Time,Type,Name,Emoji,Category,Amount,Currency,Local amount,Local currency,Notes and #tags,Address,Receipt,Description,Category split
+tx_0000AhmcbFKVlkAFbeZuAk,11/05/2024,13:41:47,Pot transfer,Savings Pot,,Savings,-1.4,GBP,-1.4,GBP,,,,Round up,
+tx_0000AhmgdZ0FzkVZou77Ev,11/05/2024,14:27:01,Card payment,D,💊,C,-100.42,GBP,-100.42,GBP,,B,,A,"Groceries:15.00,Personal care:10.00"

fixtures/transactions.json

@@ -0,0 +1,38 @@
+[
+  [
+    "tx_0000AhmcbFKVlkAFbeZuAk",
+    "2024-05-11T00:00:00.000Z",
+    "1899-12-30T13:41:47.000Z",
+    "Pot transfer",
+    "Savings Pot",
+    "",
+    "Savings",
+    -1.4,
+    "GBP",
+    -1.4,
+    "GBP",
+    "",
+    "",
+    "",
+    "Round up",
+    ""
+  ],
+  [
+    "tx_0000AhmgdZ0FzkVZou77Ev",
+    "2024-05-11T00:00:00.000Z",
+    "1899-12-30T14:27:01.000Z",
+    "Card payment",
+    "D",
+    "💊",
+    "C",
+    -100.42,
+    "GBP",
+    -100.42,
+    "GBP",
+    "",
+    "B",
+    "",
+    "A",
+    "Groceries:15.00,Personal care:10.00"
+  ]
+]

justfile

@@ -0,0 +1,12 @@
+#!/usr/bin/env just --justfile
+
+generate-migration:
+     sea-orm-cli migrate generate $1
+
+migrate:
+    sea-orm-cli migrate up
+
+generate-entities:
+    sea-orm-cli generate entity --with-serde both \
+                                -l -o entity/src \
+                                --ignore-tables monzo_ingestion_seaql_migrations

migration/Cargo.toml

@@ -12,7 +12,7 @@ path = "src/lib.rs"
 async-std = { version = "1", features = ["attributes", "tokio1"] }
 
 [dependencies.sea-orm-migration]
-version = "1.0.0-rc.4"
+version = "1.1.0"
 features = [
    "runtime-tokio-rustls",  # `ASYNC_RUNTIME` feature
    "sqlx-postgres",         # `DATABASE_DRIVER` feature

migration/src/lib.rs

@@ -1,6 +1,11 @@
 pub use sea_orm_migration::prelude::*;
 
 pub mod m20230904_141851_create_monzo_tables;
+mod m20240529_195030_add_transaction_identity_hash;
+mod m20240603_162500_make_title_optional;
+mod m20241015_195220_add_account_to_transactions;
+mod m20241015_200222_add_expenditure_transaction_fk;
+mod m20241015_200652_add_accounts;
 
 pub struct Migrator;
 
@@ -11,6 +16,13 @@ impl MigratorTrait for Migrator {
     }
 
     fn migrations() -> Vec<Box<dyn MigrationTrait>> {
-        vec![Box::new(m20230904_141851_create_monzo_tables::Migration)]
+        vec![
+            Box::new(m20230904_141851_create_monzo_tables::Migration),
+            Box::new(m20240529_195030_add_transaction_identity_hash::Migration),
+            Box::new(m20240603_162500_make_title_optional::Migration),
+            Box::new(m20241015_195220_add_account_to_transactions::Migration),
+            Box::new(m20241015_200222_add_expenditure_transaction_fk::Migration),
+            Box::new(m20241015_200652_add_accounts::Migration),
+        ]
     }
 }

migration/src/m20240529_195030_add_transaction_identity_hash.rs

@@ -0,0 +1,39 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                TableAlterStatement::new()
+                    .table(Transaction::Table)
+                    .add_column(
+                        ColumnDef::new(Transaction::IdentityHash)
+                            .big_integer()
+                            .unique_key(),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                TableAlterStatement::new()
+                    .table(Transaction::Table)
+                    .drop_column(Transaction::IdentityHash)
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum Transaction {
+    Table,
+    IdentityHash,
+}

migration/src/m20240603_162500_make_title_optional.rs

@@ -0,0 +1,61 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                TableAlterStatement::new()
+                    .table(Transaction::Table)
+                    .modify_column(ColumnDef::new(Transaction::Title).string().null())
+                    .to_owned(),
+            )
+            .await?;
+
+        // Set all empty string titles to null
+        manager
+            .get_connection()
+            .execute_unprepared(
+                r#"
+            update transaction
+            set title = null
+            where title = ''
+        "#,
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // Set all null titles to empty string when reverting
+        manager
+            .get_connection()
+            .execute_unprepared(
+                r#"
+            update transaction
+            set title = ''
+            where title is null
+        "#,
+            )
+            .await?;
+
+        manager
+            .alter_table(
+                TableAlterStatement::new()
+                    .table(Transaction::Table)
+                    .modify_column(ColumnDef::new(Transaction::Title).string().not_null())
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum Transaction {
+    Table,
+    Title,
+}

migration/src/m20241015_195220_add_account_to_transactions.rs

@@ -0,0 +1,76 @@
+use sea_orm_migration::{prelude::*, schema::*};
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                Table::create()
+                    .table(Account::Table)
+                    .if_not_exists()
+                    .col(
+                        ColumnDef::new(Account::Id)
+                            .integer()
+                            .auto_increment()
+                            .not_null()
+                            .primary_key(),
+                    )
+                    .col(
+                        ColumnDef::new(Account::Name)
+                            .string()
+                            .not_null(),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        manager.alter_table(
+            TableAlterStatement::new()
+                .table(Transaction::Table)
+                .add_column(ColumnDef::new(Transaction::AccountId).integer())
+                .to_owned(),
+        ).await?;
+
+        manager
+            .create_foreign_key(
+                ForeignKey::create()
+                    .name("fk_transaction_account_id")
+                    .from(Transaction::Table, Transaction::AccountId)
+                    .to(Account::Table, Account::Id)
+                    .on_delete(ForeignKeyAction::Cascade)
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager.alter_table(
+            TableAlterStatement::new()
+                .table(Transaction::Table)
+                .drop_column(Transaction::AccountId)
+                .to_owned(),
+        ).await?;
+
+        manager.drop_table(Table::drop().table(Account::Table).to_owned()).await?;
+
+        Ok(())
+    }
+}
+
+#[derive(DeriveIden)]
+enum Account {
+    Table,
+    Id,
+    Name,
+}
+
+#[derive(DeriveIden)]
+enum Transaction {
+    Table,
+    AccountId,
+}

migration/src/m20241015_200222_add_expenditure_transaction_fk.rs

@@ -0,0 +1,40 @@
+use sea_orm_migration::{prelude::*, schema::*};
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_foreign_key(
+                ForeignKey::create()
+                    .name("fk_expenditure_transaction_id")
+                    .from(Expenditure::Table, Expenditure::TransactionId)
+                    .to(Transaction::Table, Transaction::Id)
+                    .on_delete(ForeignKeyAction::Cascade)
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager.drop_foreign_key(ForeignKey::drop().name("fk_expenditure_transaction_id").to_owned()).await?;
+
+        Ok(())
+    }
+}
+
+#[derive(DeriveIden)]
+enum Transaction {
+    Table,
+    Id,
+}
+
+#[derive(DeriveIden)]
+enum Expenditure {
+    Table,
+    TransactionId,
+}

migration/src/m20241015_200652_add_accounts.rs

@@ -0,0 +1,53 @@
+use sea_orm_migration::{prelude::*, schema::*};
+use crate::sea_orm::sqlx;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager.exec_stmt(
+            InsertStatement::new()
+                .into_table(Account::Table)
+                .columns(vec![Account::Name])
+                .values_panic(["Monzo".into()])
+                .values_panic(["Flex".into()])
+                .to_owned()
+        ).await?;
+
+        let id: i32 = manager.get_connection().query_one(
+            sea_orm::Statement::from_string(
+                manager.get_database_backend(),
+                "SELECT id FROM account WHERE name = 'Monzo'",
+            )
+        ).await?.expect("Monzo account not found").try_get_by_index(0)?;
+
+        manager.exec_stmt(
+            UpdateStatement::new()
+                .table(Transaction::Table)
+                .values([(Transaction::AccountId, id.into())])
+                .to_owned()
+        ).await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // This is a data creation migration, so we do not roll back the migration
+        Ok(())
+    }
+}
+
+#[derive(DeriveIden)]
+enum Account {
+    Table,
+    Id,
+    Name,
+}
+
+#[derive(DeriveIden)]
+enum Transaction {
+    Table,
+    AccountId,
+}

src/error.rs

@@ -6,25 +6,33 @@ use tracing::log::error;
 #[derive(thiserror::Error, Debug)]
 pub enum AppError {
     /// SeaORM error, separated for ease of use allowing us to `?` db operations.
-    #[error("Internal error")]
+    #[error("Database error: {0}")]
     DbError(#[from] DbErr),
 
     #[error("Invalid request {0}")]
     BadRequest(anyhow::Error),
 
     /// Catch all for error we don't care to expose publicly.
-    #[error("Internal error")]
+    #[error("An error occurred: {0}")]
     Anyhow(#[from] anyhow::Error),
 }
 
+impl AppError {
+    fn to_response_string(&self) -> String {
+        match self {
+            AppError::BadRequest(e) => e.to_string(),
+            _ => "Internal server error".to_string(),
+        }
+    }
+}
+
 impl IntoResponse for AppError {
     fn into_response(self) -> Response {
-        error!("Internal server error: {self:?}");
-
         let status_code = match self {
+            AppError::BadRequest(_) => StatusCode::BAD_REQUEST,
             _ => StatusCode::INTERNAL_SERVER_ERROR,
         };
 
-        (status_code, self.to_string()).into_response()
+        (status_code, self.to_response_string()).into_response()
     }
 }

src/ingestion/db.rs

@@ -1,34 +1,75 @@
 use crate::error::AppError;
+use crate::ingestion::ingestion_logic::MonzoRow;
 use anyhow::anyhow;
 use entity::{expenditure, transaction};
-use sea_orm::sea_query::{OnConflict, PostgresQueryBuilder};
+use sea_orm::sea_query::OnConflict;
-use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, Iterable, TransactionTrait};
 use sea_orm::{
-    ConnectionTrait, DatabaseBackend, DatabaseTransaction, DbErr, QueryFilter, QueryTrait,
+    ColumnTrait, DatabaseConnection, DbErr, EntityTrait, Iterable, QuerySelect, TransactionTrait,
-    Statement,
 };
+use sea_orm::{ConnectionTrait, DatabaseTransaction, QueryFilter};
 
 pub struct Insertion {
     pub transaction: transaction::ActiveModel,
     pub contained_expenditures: Vec<expenditure::ActiveModel>,
+    pub identity_hash: i64,
+}
+
+// Note while this is more efficient in db calls, it does bind together the entire group.
+// We employ a batching process for now to try balance speed and failure rate, but it is worth
+// trying to move failures earlier and improve reporting.
+pub async fn insert(
+    db: &DatabaseConnection,
+    monzo_rows: Vec<MonzoRow>,
+    account_id: i32,
+) -> Result<Vec<String>, AppError> {
+    let mut new_transaction_ids = Vec::new();
+    let insertions = monzo_rows
+        .into_iter()
+        .map(|row| MonzoRow::into_insertion(row, account_id))
+        .collect::<Result<Vec<_>, _>>()?;
+
+    for insertions in insertions.chunks(200) {
+        let (new_or_updated_insertions, inserted_transaction_ids) =
+            whittle_insertions(insertions, db).await?;
+
+        if new_or_updated_insertions.is_empty() {
+            continue;
+        }
+
+        let tx = db.begin().await?;
+        update_transactions(&tx, &new_or_updated_insertions).await?;
+        update_expenditures(&tx, &new_or_updated_insertions, &inserted_transaction_ids).await?;
+        tx.commit().await?;
+
+        // We wait until the transaction is committed before adding the new transaction ids to the
+        // list to avoid issues with the transaction being rolled back.
+        new_transaction_ids.extend(inserted_transaction_ids);
+    }
+
+    // Notify the new transactions once everything is committed.
+    notify_new_transactions(db, &new_transaction_ids).await?;
+
+    Ok(new_transaction_ids)
 }
 
 async fn update_expenditures(
     tx: &DatabaseTransaction,
-    insertions: &[Insertion],
+    new_or_updated_insertions: &[&Insertion],
-) -> Result<(), DbErr> {
+    inserted_transaction_ids: &[String],
+) -> Result<(), AppError> {
+    if new_or_updated_insertions.is_empty() {
+        return Ok(());
+    }
+
     // Expenditures can change as we re-categorise them, so we delete all the old ones and
     // insert an entirely new set to ensure we don't end up leaving old ones around.
     expenditure::Entity::delete_many()
-        .filter(
+        .filter(expenditure::Column::TransactionId.is_in(inserted_transaction_ids))
-            expenditure::Column::TransactionId
-                .is_in(insertions.iter().map(|i| i.transaction.id.as_ref())),
-        )
         .exec(tx)
         .await?;
 
     expenditure::Entity::insert_many(
-        insertions
+        new_or_updated_insertions
             .iter()
             .flat_map(|i| &i.contained_expenditures)
             .cloned(),
@@ -43,62 +84,65 @@ async fn update_expenditures(
     )
     .exec(tx)
     .await?;
-
     Ok(())
 }
 
-// Note while this is more efficient in db calls, it does bind together the entire group.
+async fn update_transactions(
-// We employ a batching process for now to try balance speed and failure rate, but it is worth
+    tx: &DatabaseTransaction,
-// trying to move failures earlier and improve reporting.
+    new_or_updated_insertions: &[&Insertion],
-pub async fn insert(
+) -> Result<(), DbErr> {
-    db: &DatabaseConnection,
+    if new_or_updated_insertions.is_empty() {
-    insertions: Vec<Insertion>,
+        return Ok(());
-) -> Result<Vec<String>, AppError> {
-    let mut new_transaction_ids = Vec::new();
-
-    for insertions in insertions.chunks(400) {
-        let tx = db.begin().await?;
-        let inserted_transaction_ids = update_transactions(insertions, &tx).await?;
-        update_expenditures(&tx, &insertions).await?;
-        tx.commit().await?;
-
-        // We wait until the transaction is committed before adding the new transaction ids to the
-        // list to avoid issues with the transaction being rolled back.
-        new_transaction_ids.extend(inserted_transaction_ids);
     }
 
-    // Notify the new transactions once everything is committed.
+    let transactions = new_or_updated_insertions
-    notify_new_transactions(db, &new_transaction_ids).await?;
+        .iter()
+        .map(|i| &i.transaction)
+        .cloned();
 
-    Ok(new_transaction_ids)
+    transaction::Entity::insert_many(transactions)
+        .on_conflict(
+            OnConflict::column(transaction::Column::Id)
+                .update_columns(transaction::Column::iter())
+                .to_owned(),
+        )
+        .exec(tx)
+        .await?;
+    Ok(())
 }
 
-async fn update_transactions(
+async fn whittle_insertions<'a>(
-    insertions: &[Insertion],
+    insertions: &'a [Insertion],
-    tx: &DatabaseTransaction,
+    tx: &DatabaseConnection,
-) -> Result<Vec<String>, AppError> {
+) -> Result<(Vec<&'a Insertion>, Vec<String>), AppError> {
-    let insert =
+    let existing_hashes = transaction::Entity::find()
-        transaction::Entity::insert_many(insertions.iter().map(|i| &i.transaction).cloned())
+        .select_only()
-            .on_conflict(
+        .columns([transaction::Column::IdentityHash])
-                OnConflict::column(transaction::Column::Id)
+        .filter(transaction::Column::IdentityHash.is_not_null())
-                    .update_columns(transaction::Column::iter())
+        .into_tuple::<(i64,)>()
-                    .to_owned(),
+        .all(tx)
-            )
+        .await?;
-            .into_query()
-            .returning_col(transaction::Column::Id)
-            .build(PostgresQueryBuilder);
 
-    let inserted_transaction_ids = tx
+    tracing::debug!("Found existing entries: {existing_hashes:?}");
-        .query_all(Statement::from_sql_and_values(
+
-            DatabaseBackend::Postgres,
+    // We will only update those where the hash is different to avoid unnecessary updates and
-            insert.0,
+    // notifications.
-            insert.1,
+    let new_or_updated_insertions = insertions
-        ))
-        .await?
         .iter()
-        .map(|r| r.try_get_by("id"))
+        .filter(|i| {
-        .collect::<Result<Vec<String>, _>>()?;
+            let hash = i.identity_hash;
-    Ok(inserted_transaction_ids)
+            !existing_hashes
+                .iter()
+                .any(|(existing_hash,)| *existing_hash == hash)
+        })
+        .collect::<Vec<_>>();
+
+    let inserted_transaction_ids = new_or_updated_insertions
+        .iter()
+        .map(|i| i.transaction.id.clone().unwrap())
+        .collect::<Vec<_>>();
+
+    Ok((new_or_updated_insertions, inserted_transaction_ids))
 }
 
 async fn notify_new_transactions(
@@ -114,16 +158,29 @@ async fn notify_new_transactions(
 }
 
 mod tests {
+    use super::{insert, notify_new_transactions, update_expenditures, update_transactions};
+    use crate::ingestion::ingestion_logic::from_json_row;
     use anyhow::Error;
-    use sea_orm::{DatabaseConnection};
+    use entity::account;
-    use sqlx::{PgPool};
-    use sqlx::postgres::PgListener;
-    use testcontainers::ContainerAsync;
     use migration::MigratorTrait;
+    use sea_orm::{ActiveModelTrait, DatabaseConnection, TransactionTrait};
+    use serde_json::Value;
+    use sqlx::postgres::PgListener;
+    use sqlx::{Executor, PgPool};
     use testcontainers::runners::AsyncRunner;
-    use super::notify_new_transactions;
+    use testcontainers::ContainerAsync;
+    use tokio::sync::OnceCell;
 
-    async fn initialise() -> Result<(ContainerAsync<testcontainers_modules::postgres::Postgres>, DatabaseConnection, PgPool), Error> {
+    #[derive(Debug)]
+    struct DatabaseInstance {
+        container: ContainerAsync<testcontainers_modules::postgres::Postgres>,
+        db: DatabaseConnection,
+        pool: PgPool,
+    }
+
+    static INSTANCE: OnceCell<DatabaseInstance> = OnceCell::const_new();
+
+    async fn initialise_db() -> Result<DatabaseInstance, Error> {
         let container = testcontainers_modules::postgres::Postgres::default()
             .start()
             .await?;
@@ -137,30 +194,143 @@ mod tests {
         let db: DatabaseConnection = sea_orm::Database::connect(connection_string).await?;
         migration::Migrator::up(&db, None).await?;
 
-        let pool = PgPool::connect(connection_string)
+        let pool = PgPool::connect(connection_string).await?;
-            .await?;
+        let instance = DatabaseInstance {
+            container,
+            db,
+            pool,
+        };
 
-        Ok((container, db, pool))
+        Ok(instance)
+    }
+
+    async fn get_or_initialize_db_instance() -> Result<&'static DatabaseInstance, Error> {
+        Ok(INSTANCE
+            .get_or_init(|| async { initialise_db().await.unwrap() })
+            .await)
+    }
+
+    async fn create_test_account(db: &DatabaseConnection) -> Result<i32, Error> {
+        let new_account = account::ActiveModel {
+            id: sea_orm::ActiveValue::NotSet,
+            name: sea_orm::ActiveValue::Set("Test Account".to_string()),
+        };
+        let inserted = new_account.insert(db).await?;
+        Ok(inserted.id)
+    }
+
+    #[tokio::test]
+    async fn test_empty_insertion_list() -> Result<(), Error> {
+        let db = get_or_initialize_db_instance().await?;
+        let insertions = vec![];
+        let tx = db.db.begin().await?;
+        update_transactions(&tx, &insertions).await?;
+        update_expenditures(&tx, &insertions, &vec![]).await?;
+        tx.commit().await?;
+
+        Ok(())
     }
 
     #[tokio::test]
     async fn test_notify() -> Result<(), Error> {
-        let (_container, db, pool) = initialise().await?;
+        let dbi = get_or_initialize_db_instance().await?;
-        let mut listener = PgListener::connect_with(&pool).await?;
+        let mut listener = PgListener::connect_with(&dbi.pool).await?;
         listener.listen("monzo_new_transactions").await?;
 
-        let ids = vec!["test1".to_string(), "test2".to_string(), "test3".to_string()];
+        let ids = vec![
+            "test1".to_string(),
+            "test2".to_string(),
+            "test3".to_string(),
+        ];
 
-        notify_new_transactions(
+        notify_new_transactions(&dbi.db, &ids).await?;
-            &db,
-            &ids,
-        ).await?;
 
         let notification = listener.recv().await?;
         let payload = notification.payload();
         println!("Payload: {}", payload);
-        assert_eq!(serde_json::from_str::<Vec<String>>(&payload)?, ids, "Payloads do not match");
+        assert_eq!(
+            serde_json::from_str::<Vec<String>>(&payload)?,
+            ids,
+            "Payloads do not match"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_notify_on_insert() -> Result<(), Error> {
+        let dbi = get_or_initialize_db_instance().await?;
+        let account_id = create_test_account(&dbi.db).await?;
+        let mut listener = PgListener::connect_with(&dbi.pool).await?;
+        listener.listen("monzo_new_transactions").await?;
+
+        let json = include_str!("../../fixtures/transactions.json");
+        let json: Vec<Vec<Value>> = serde_json::from_str(json).unwrap();
+        let data = json
+            .iter()
+            .map(|row| from_json_row(row))
+            .collect::<Result<Vec<_>, anyhow::Error>>()?;
+
+        insert(&dbi.db, data.clone(), account_id).await?;
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        let mut payload = serde_json::from_str::<Vec<String>>(&payload)?;
+        payload.sort();
+
+        let mut ids = data
+            .iter()
+            .map(|row| row.transaction_id.clone())
+            .collect::<Vec<_>>();
+
+        ids.sort();
+
+        assert_eq!(payload, ids, "Inserted IDs do not match");
+
+        insert(&dbi.db, data.clone(), account_id).await?;
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        let payload = serde_json::from_str::<Vec<String>>(&payload)?;
+        assert_eq!(
+            payload,
+            Vec::<String>::new(),
+            "Re-inserting identical rows triggered double notification"
+        );
+
+        let mut altered_data = data.clone();
+        altered_data[0].description = Some("New description".to_string());
+
+        assert_ne!(
+            altered_data[0].compute_hash(),
+            data[0].compute_hash(),
+            "Alterations have the same hash"
+        );
+
+        insert(&dbi.db, altered_data.clone(), account_id).await?;
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        let payload = serde_json::from_str::<Vec<String>>(&payload)?;
+        assert_eq!(
+            payload,
+            vec![altered_data[0].transaction_id.clone()],
+            "Re-inserting altered row failed to re-trigger notification"
+        );
 
         Ok(())
     }
 }
+
+pub(crate) async fn get_account_id(
+    p0: &DatabaseConnection,
+    p1: Option<String>,
+) -> Result<i32, AppError> {
+    let p1 = p1.unwrap_or("Monzo".to_string());
+
+    entity::prelude::Account::find()
+        .filter(entity::account::Column::Name.eq(p1))
+        .select_only()
+        .column(entity::account::Column::Id)
+        .into_tuple::<i32>()
+        .one(p0)
+        .await?
+        .ok_or(AppError::BadRequest(anyhow!("Account not found")))
+}

src/ingestion/flex.rs

@@ -0,0 +1,9 @@
+#[allow(dead_code)]
+pub mod headings {
+    #[allow(unused_imports)]
+    pub use super::super::ingestion_logic::headings::*;
+
+    // Additional Flex headings
+    pub const MONEY_OUT: usize = 16;
+    pub const MONEY_IN: usize = 17;
+}

src/ingestion/ingestion_logic.rs

@@ -1,16 +1,18 @@
 use crate::ingestion::db::Insertion;
-use anyhow::Context;
+use anyhow::{anyhow, Context};
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime};
 use csv::StringRecord;
 use entity::expenditure::ActiveModel;
 use entity::transaction;
 use num_traits::FromPrimitive;
 use sea_orm::prelude::Decimal;
-use sea_orm::ActiveValue::*;
 use sea_orm::IntoActiveModel;
+use serde_json::Value;
+use std::hash::Hash;
+use crate::ingestion::flex;
 
 #[allow(dead_code)]
-mod headings {
+pub(crate) mod headings {
     pub const TRANSACTION_ID: usize = 0;
     pub const DATE: usize = 1;
     pub const TIME: usize = 2;
@@ -29,41 +31,116 @@ mod headings {
     pub const CATEGORY_SPLIT: usize = 15;
 }
 
-fn parse_section(monzo_transaction_id: &str, section: &str) -> anyhow::Result<ActiveModel> {
+#[derive(Debug, Eq, PartialEq, Hash, Clone)]
-    let mut components = section.split(':');
+pub struct MonzoRow {
-    let category: String = components
+    pub category_split: Option<String>,
-        .next()
+    pub primary_category: String,
-        .context("Missing Missing category")?
+    pub total_amount: Decimal,
-        .to_string();
+    pub receipt: Option<String>,
-
+    pub notes: Option<String>,
-    let amount = components
+    pub emoji: Option<String>,
-        .next()
+    pub description: Option<String>,
-        .context("Missing amount")?
+    pub transaction_type: String,
-        .parse::<Decimal>()?;
+    pub title: Option<String>,
-
+    pub timestamp: NaiveDateTime,
-    Ok(entity::expenditure::Model {
+    pub transaction_id: String,
-        transaction_id: monzo_transaction_id.to_string(),
-        category,
-        amount,
-    }
-    .into_active_model())
 }
 
-fn json_opt(value: &serde_json::Value) -> Option<String> {
+impl MonzoRow {
+    fn parse_section(monzo_transaction_id: &str, section: &str) -> anyhow::Result<ActiveModel> {
+        let mut components = section.split(':');
+        let category: String = components
+            .next()
+            .context("Missing Missing category")?
+            .to_string();
+
+        let amount = components
+            .next()
+            .context("Missing amount")?
+            .parse::<Decimal>()?;
+
+        Ok(entity::expenditure::Model {
+            transaction_id: monzo_transaction_id.to_string(),
+            category,
+            amount,
+        }
+        .into_active_model())
+    }
+
+    /// Compute a hash of this row, returning the number as an i64 to be used as a unique constraint
+    /// in the database.
+    pub fn compute_hash(&self) -> i64 {
+        use std::collections::hash_map::DefaultHasher;
+        use std::hash::Hasher;
+
+        let mut hasher = DefaultHasher::new();
+        self.hash(&mut hasher);
+        hasher.finish() as i64
+    }
+
+    pub fn into_insertion(self, account_id: i32) -> Result<Insertion, anyhow::Error> {
+        let identity_hash = self.compute_hash();
+
+        let expenditures: Vec<_> = match &self.category_split {
+            Some(split) if !split.is_empty() => split
+                .split(',')
+                .map(|section| Self::parse_section(&self.transaction_id, section))
+                .collect::<Result<Vec<_>, anyhow::Error>>()?,
+
+            _ => vec![entity::expenditure::Model {
+                category: self.primary_category.clone(),
+                amount: self.total_amount,
+                transaction_id: self.transaction_id.clone(),
+            }
+            .into_active_model()],
+        };
+
+        Ok(Insertion {
+            transaction: transaction::Model {
+                id: self.transaction_id,
+                transaction_type: self.transaction_type,
+                timestamp: self.timestamp,
+                title: self.title,
+                emoji: self.emoji,
+                notes: self.notes,
+                receipt: self.receipt,
+                total_amount: self.total_amount,
+                description: self.description,
+                identity_hash: Some(identity_hash),
+                account_id: Some(account_id),
+            }
+            .into_active_model(),
+
+            contained_expenditures: expenditures,
+            identity_hash,
+        })
+    }
+}
+
+fn json_opt(value: &Value) -> Option<String> {
     match value {
-        serde_json::Value::String(string) if string.is_empty() => None,
+        Value::String(string) if string.is_empty() => None,
-        serde_json::Value::String(string) => Some(string.to_string()),
+        Value::String(string) => Some(string.to_string()),
         _ => None,
     }
 }
 
-pub fn from_json_row(row: Vec<serde_json::Value>) -> anyhow::Result<Insertion> {
+fn json_required_str(value: &Value, label: &str) -> anyhow::Result<String> {
-    use serde_json::Value;
+    match value {
-    let monzo_transaction_id = row[headings::TRANSACTION_ID]
+        Value::String(string) if string.is_empty() => Err(anyhow!("{} is empty", label)),
-        .as_str()
+        Value::String(string) => Ok(string.to_string()),
-        .context("No transaction id")?
+        _ => Err(anyhow!("{} is not a string", label)),
-        .to_string();
+    }
+}
 
+fn parse_timestamp(date: &str, time: &str) -> anyhow::Result<NaiveDateTime> {
+    let date = NaiveDate::parse_from_str(date, "%Y-%m-%d")?;
+    let time = NaiveTime::parse_from_str(time, "%H:%M:%S")?;
+
+    Ok(date.and_time(time))
+}
+
+pub fn from_json_row(row: &[Value]) -> anyhow::Result<MonzoRow> {
     let date = DateTime::parse_from_rfc3339(row[headings::DATE].as_str().context("No date")?)
         .context("Failed to parse date")?;
 
@@ -73,57 +150,57 @@ pub fn from_json_row(row: Vec<serde_json::Value>) -> anyhow::Result<Insertion> {
 
     let timestamp = date.date_naive().and_time(time);
 
-    let title = row[headings::NAME]
-        .as_str()
-        .context("No title")?
-        .to_string();
-
-    let monzo_transaction_type = row[headings::TYPE]
-        .as_str()
-        .context("No transaction type")?
-        .to_string();
-
-    let description = json_opt(&row[headings::DESCRIPTION]);
-    let emoji = json_opt(&row[headings::EMOJI]);
-    let notes = json_opt(&row[headings::NOTES_AND_TAGS]);
-    let receipt = json_opt(&row[headings::RECEIPT]);
     let total_amount = Decimal::from_f64(row[headings::AMOUNT].as_f64().context("No amount")?)
         .context("Failed to parse date")?;
 
-    let expenditures: Vec<_> = match row.get(headings::CATEGORY_SPLIT) {
+    Ok(MonzoRow {
-        Some(Value::String(split)) if !split.is_empty() => split
+        transaction_id: json_required_str(&row[headings::TRANSACTION_ID], "Transaction ID")?,
-            .split(',')
+        title: json_opt(&row[headings::NAME]),
-            .map(|section| parse_section(&monzo_transaction_id, section))
+        transaction_type: json_required_str(&row[headings::TYPE], "Transaction type")?,
-            .collect::<Result<Vec<_>, anyhow::Error>>()?,
+        description: json_opt(&row[headings::DESCRIPTION]),
-
+        emoji: json_opt(&row[headings::EMOJI]),
-        _ => vec![entity::expenditure::Model {
+        notes: json_opt(&row[headings::NOTES_AND_TAGS]),
-            category: row[headings::CATEGORY]
+        receipt: json_opt(&row[headings::RECEIPT]),
-                .as_str()
+        primary_category: json_required_str(&row[headings::CATEGORY], "Primary Category")?,
-                .context("No context")?
+        category_split: json_opt(&row[headings::CATEGORY_SPLIT]),
-                .to_string(),
+        total_amount,
-            amount: total_amount,
+        timestamp,
-            transaction_id: monzo_transaction_id.clone(),
-        }
-        .into_active_model()],
-    };
-
-    Ok(Insertion {
-        transaction: transaction::ActiveModel {
-            id: Set(monzo_transaction_id),
-            transaction_type: Set(monzo_transaction_type),
-            timestamp: Set(timestamp),
-            title: Set(title),
-            emoji: Set(emoji),
-            notes: Set(notes),
-            receipt: Set(receipt),
-            total_amount: Set(total_amount),
-            description: Set(description),
-        },
-
-        contained_expenditures: expenditures,
     })
 }
 
+#[test]
+fn test_json() {
+    let json = include_str!("../../fixtures/transactions.json");
+    let csv = include_str!("../../fixtures/transactions.csv");
+
+    let json: Vec<Vec<Value>> = serde_json::from_str(json).unwrap();
+    let mut csv_reader = csv::Reader::from_reader(csv.as_bytes());
+
+    let json_rows = json
+        .iter()
+        .map(|row| from_json_row(&row))
+        .collect::<Result<Vec<_>, anyhow::Error>>()
+        .unwrap();
+
+    let csv_rows = csv_reader
+        .records()
+        .map(|record| from_csv_row(record.unwrap()))
+        .collect::<Result<Vec<_>, anyhow::Error>>()
+        .unwrap();
+
+    assert_eq!(csv_rows.len(), json_rows.len(), "Different number of rows");
+
+    for (i, (json_row, csv_row)) in json_rows.iter().zip(csv_rows.iter()).enumerate() {
+        assert_eq!(json_row, csv_row, "Row {} is different", i);
+        assert_eq!(
+            json_row.compute_hash(),
+            csv_row.compute_hash(),
+            "Row {} hash are different",
+            i
+        );
+    }
+}
+
 fn csv_opt(s: &str) -> Option<String> {
     match s {
         "" => None,
@@ -131,9 +208,7 @@ fn csv_opt(s: &str) -> Option<String> {
     }
 }
 
-pub fn from_csv_row(row: StringRecord) -> anyhow::Result<Insertion> {
+pub fn from_csv_row(row: StringRecord) -> anyhow::Result<MonzoRow> {
-    let monzo_transaction_id = row[headings::TRANSACTION_ID].to_string();
-
     let date = NaiveDate::parse_from_str(&row[headings::DATE], "%d/%m/%Y")
         .context("Failed to parse date from csv")?;
 
@@ -142,41 +217,17 @@ pub fn from_csv_row(row: StringRecord) -> anyhow::Result<Insertion> {
 
     let timestamp = NaiveDateTime::new(date, time);
 
-    let title = row[headings::NAME].to_string();
+    Ok(MonzoRow {
-    let monzo_transaction_type = row[headings::TYPE].to_string();
+        timestamp,
-
+        transaction_id: row[headings::TRANSACTION_ID].to_string(),
-    let description = csv_opt(&row[headings::DESCRIPTION]);
+        title: csv_opt(&row[headings::NAME]),
-    let emoji = csv_opt(&row[headings::EMOJI]);
+        transaction_type: row[headings::TYPE].to_string(),
-    let notes = csv_opt(&row[headings::NOTES_AND_TAGS]);
+        description: csv_opt(&row[headings::DESCRIPTION]),
-    let receipt = csv_opt(&row[headings::RECEIPT]);
+        emoji: csv_opt(&row[headings::EMOJI]),
-    let total_amount = row[headings::AMOUNT].parse::<Decimal>()?;
+        notes: csv_opt(&row[headings::NOTES_AND_TAGS]),
-
+        receipt: csv_opt(&row[headings::RECEIPT]),
-    let expenditures: Vec<_> = match row.get(headings::CATEGORY_SPLIT) {
+        total_amount: row[headings::AMOUNT].parse::<Decimal>()?,
-        Some(split) if !split.is_empty() => split
+        category_split: csv_opt(&row[headings::CATEGORY_SPLIT]),
-            .split(',')
+        primary_category: row[headings::CATEGORY].to_string(),
-            .map(|section| parse_section(&monzo_transaction_id, section))
-            .collect::<Result<Vec<_>, anyhow::Error>>()?,
-
-        _ => vec![entity::expenditure::Model {
-            transaction_id: monzo_transaction_id.clone(),
-            category: row[headings::CATEGORY].to_string(),
-            amount: total_amount,
-        }
-        .into_active_model()],
-    };
-
-    Ok(Insertion {
-        transaction: transaction::ActiveModel {
-            id: Set(monzo_transaction_id),
-            transaction_type: Set(monzo_transaction_type),
-            timestamp: Set(timestamp),
-            title: Set(title),
-            emoji: Set(emoji),
-            notes: Set(notes),
-            receipt: Set(receipt),
-            total_amount: Set(total_amount),
-            description: Set(description),
-        },
-        contained_expenditures: expenditures,
     })
 }

src/ingestion/mod.rs

@@ -1,3 +1,4 @@
 pub mod db;
 pub mod ingestion_logic;
+pub mod flex;
 pub mod routes;

src/ingestion/routes.rs

@@ -2,66 +2,133 @@ use crate::error::AppError;
 use crate::ingestion::db;
 use crate::ingestion::ingestion_logic::{from_csv_row, from_json_row};
 use anyhow::anyhow;
+use axum::extract::multipart::MultipartError;
 use axum::extract::{Extension, Json, Multipart};
+use bytes::Bytes;
 use sea_orm::DatabaseConnection;
 use serde_json::Value;
 use std::io::Cursor;
 
-pub async fn monzo_updated(
+#[derive(serde::Deserialize, Debug)]
-    Extension(db): Extension<DatabaseConnection>,
+#[serde(untagged)]
-    Json(row): Json<Vec<Value>>,
+pub enum MonzoBatchedJsonInput {
-) -> Result<&'static str, AppError> {
+    Legacy(Vec<Vec<Value>>),
-    db::insert(&db, vec![from_json_row(row)?]).await.unwrap();
+    New {
+        account_id: Option<u8>,
+        rows: Vec<Vec<Value>>,
+    },
+}
 
-    Ok("Ok")
+impl MonzoBatchedJsonInput {
+    fn account_id(&self) -> Option<u8> {
+        match self {
+            MonzoBatchedJsonInput::Legacy(_) => None,
+            MonzoBatchedJsonInput::New { account_id, .. } => *account_id,
+        }
+    }
+
+    fn rows(&self) -> &[Vec<Value>] {
+        match self {
+            MonzoBatchedJsonInput::Legacy(rows) => rows,
+            MonzoBatchedJsonInput::New { rows, .. } => rows,
+        }
+    }
 }
 
 pub async fn monzo_batched_json(
     Extension(db): Extension<DatabaseConnection>,
-    Json(data): Json<Vec<Vec<Value>>>,
+    Json(data): Json<MonzoBatchedJsonInput>,
 ) -> Result<&'static str, AppError> {
-    let insertions = data
+    let rows = data
-        .into_iter()
+        .rows()
-        .skip(1)
+        .iter()
-        .map(|row| from_json_row(row))
+        .skip_while(|row| row[0] == Value::String("Transaction ID".to_string()))
+        .map(|row| from_json_row(row.as_ref()))
         .collect::<Result<_, _>>()?;
 
-    db::insert(&db, insertions).await.unwrap();
+    // We default to the main account for JSON ingestion for now.
+    db::insert(&db, rows, data.account_id().unwrap_or(1) as i32).await?;
+
+    Ok("Ok")
+}
+
+async fn extract_csv_and_account_name(
+    mut multipart: Multipart,
+) -> Result<(Option<Bytes>, Option<String>), MultipartError> {
+    let mut csv = None;
+    let mut account_name = None;
+
+    while let Some(field) = multipart.next_field().await? {
+        match field.name() {
+            Some("csv") => {
+                csv = Some(field.bytes().await?);
+            }
+
+            Some("account_id") => {
+                account_name = Some(field.text().await?);
+            }
+
+            _ => {}
+        }
+
+        if csv.is_some() && account_name.is_some() {
+            break;
+        }
+    }
+
+    Ok((csv, account_name))
+}
+
+#[derive(serde::Deserialize, Debug, Clone)]
+pub struct ShortcutBody {
+    pub body: String,
+    pub account_name: String,
+}
+
+pub async fn shortcuts_csv(
+    Extension(db): Extension<DatabaseConnection>,
+    Json(shortcut_body): Json<ShortcutBody>,
+) -> Result<&'static str, AppError> {
+    let account_id = db::get_account_id(&db, Some(shortcut_body.account_name)).await?;
+
+    let csv = Cursor::new(shortcut_body.body.as_bytes());
+    let mut csv = csv::Reader::from_reader(csv);
+    let data = csv.records();
+    let data = data
+        .filter_map(|f| f.ok())
+        .map(from_csv_row)
+        .collect::<Result<_, _>>()?;
+
+    db::insert(&db, data, account_id).await?;
 
     Ok("Ok")
 }
 
 pub async fn monzo_batched_csv(
     Extension(db): Extension<DatabaseConnection>,
-    mut multipart: Multipart,
+    multipart: Multipart,
 ) -> Result<&'static str, AppError> {
-    let csv = loop {
+    static CSV_MISSING_ERR_MSG: &str = "No CSV file provided. Expected a multipart request with a `csv` field containing the contents of the CSV.";
-        match multipart.next_field().await.unwrap() {
-            Some(field) if field.name() == Some("csv") => {
-                break Some(field.bytes().await.unwrap());
-            }
 
-            Some(_) => {}
+    let (csv, account_name) = extract_csv_and_account_name(multipart)
-            None => break None,
+        .await
-        }
+        .map_err(|e| AppError::BadRequest(anyhow!(e)))?;
-    };
 
     let Some(csv) = csv else {
-        return Err(AppError::BadRequest(anyhow!("No CSV file provided")));
+        return Err(AppError::BadRequest(anyhow!(CSV_MISSING_ERR_MSG)));
     };
 
+    let account_id = db::get_account_id(&db, account_name).await?;
+
     let csv = Cursor::new(csv);
     let mut csv = csv::Reader::from_reader(csv);
     let data = csv.records();
+    let data = data
+        .filter_map(|f| f.ok())
+        .map(from_csv_row)
+        .collect::<Result<_, _>>()?;
 
-    db::insert(
+    db::insert(&db, data, account_id).await?;
-        &db,
-        data.filter_map(|f| f.ok())
-            .map(from_csv_row)
-            .collect::<Result<_, _>>()?,
-    )
-    .await
-    .unwrap();
 
     Ok("Ok")
 }

src/main.rs

@@ -2,29 +2,64 @@ mod error;
 mod ingestion;
 
 use crate::error::AppError;
-use crate::ingestion::routes::{monzo_batched_csv, monzo_batched_json, monzo_updated};
+use crate::ingestion::db;
+use crate::ingestion::ingestion_logic::from_csv_row;
+use crate::ingestion::routes::{monzo_batched_csv, monzo_batched_json, shortcuts_csv};
 use axum::routing::{get, post};
 use axum::{Extension, Router};
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use migration::{Migrator, MigratorTrait};
 use sea_orm::{ConnectionTrait, DatabaseConnection};
+use std::fs::File;
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use tower_http::trace::TraceLayer;
 use tracing::log::LevelFilter;
 
+#[derive(Debug, Subcommand)]
+enum Commands {
+    /// Manually run database migrations.
+    Migrate {
+        /// Number of migration steps to perform. If not provided, all migrations will be run.
+        #[arg(long)]
+        steps: Option<u32>,
+
+        /// If we should perform migration down.
+        #[arg(long)]
+        down: bool,
+    },
+
+    /// Start web app for to the google-sheets app script
+    Serve {
+        /// If we should perform migration at startup.
+        #[clap(short, long, env, default_value_t = true)]
+        migrate: bool,
+
+        /// The server address to bind to.
+        #[clap(short, long, env, default_value = "0.0.0.0:3000")]
+        addr: SocketAddr,
+    },
+
+    /// Ingest a google-sheets CSV export into the database.
+    Csv {
+        /// The path of the CSV file to ingest.
+        csv_file: PathBuf,
+
+        /// The name of the account to ingest the CSV for.
+        #[clap(long, short)]
+        account: String,
+    },
+}
+
+/// Ingest and manage monzo transactions from the Monzo Plus auto-export feature
 #[derive(Debug, clap::Parser)]
-struct Config {
+struct Cli {
-    /// If we should perform migration on startup.
-    #[clap(short, long, env, default_value_t = true)]
-    migrate: bool,
-
-    /// The server address to bind to.
-    #[clap(short, long, env, default_value = "0.0.0.0:3000")]
-    addr: SocketAddr,
-
     /// URL to PostgreSQL database.
     #[clap(short, long = "db", env)]
     database_url: String,
+
+    #[command(subcommand)]
+    command: Commands,
 }
 
 async fn health_check(
@@ -37,29 +72,60 @@ async fn health_check(
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let config: Config = Config::parse();
+    tracing_subscriber::fmt::init();
-    let connection = sea_orm::ConnectOptions::new(&config.database_url)
+
+    let cli: Cli = Cli::parse();
+    let connection = sea_orm::ConnectOptions::new(&cli.database_url)
         .sqlx_logging_level(LevelFilter::Debug)
         .to_owned();
 
     let connection = sea_orm::Database::connect(connection).await?;
 
-    if config.migrate {
+    match cli.command {
-        Migrator::up(&connection, None).await?;
+        Commands::Migrate { steps, down } => {
+            if down {
+                Migrator::down(&connection, steps).await?;
+            } else {
+                Migrator::up(&connection, steps).await?
+            }
+        }
+
+        Commands::Serve { migrate, addr } => {
+            if migrate {
+                Migrator::up(&connection, None).await?;
+            }
+
+            serve_web(addr, connection).await?;
+        }
+
+        Commands::Csv { csv_file, account: account_name } => {
+            let mut csv = csv::Reader::from_reader(File::open(csv_file)?);
+            let data = csv.records();
+            let data = data
+                .filter_map(|f| f.ok())
+                .map(from_csv_row)
+                .collect::<Result<_, _>>()?;
+
+            let account_id = db::get_account_id(&connection, Some(account_name)).await?;
+            db::insert(&connection, data, account_id).await?;
+        }
     }
 
-    tracing_subscriber::fmt::init();
+    Ok(())
+}
+
+async fn serve_web(address: SocketAddr, connection: DatabaseConnection) -> anyhow::Result<()> {
     let app = Router::new()
         .route("/health", get(health_check))
-        .route("/monzo-updated", post(monzo_updated))
         .route("/monzo-batch-export", post(monzo_batched_json))
         .route("/monzo-csv-ingestion", post(monzo_batched_csv))
+        .route("/shortcuts-csv-import", post(shortcuts_csv))
         .layer(Extension(connection.clone()))
         .layer(TraceLayer::new_for_http());
 
-    tracing::debug!("listening on {}", &config.addr);
+    tracing::info!("listening on {}", &address);
-    let listener = tokio::net::TcpListener::bind(&config.addr).await.unwrap();
+    let listener = tokio::net::TcpListener::bind(&address).await?;
-    axum::serve(listener, app).await.unwrap();
+    axum::serve(listener, app).await?;
 
     Ok(())
 }