Fix accounts transform (#5)

* add debug configs for CLI in vscode

* only operate on latest file in transform accounts

* rewrite TransformAccounts to only process newest file

* adjust balances schema

* adjust balances schema

* fix import, net worth

---------

Co-authored-by: quinn <[email protected]>

.vscode/launch.json

@@ -27,6 +27,33 @@
 			"args": [".", "--remote-debugging-port=9222"],
 			"outputCapture": "std",
 			"console": "integratedTerminal"
+		},
+		{
+			"name": "CLI: Extract",
+			"cwd": "${workspaceFolder}",
+			"type": "go",
+			"request": "launch",
+			"program": "${workspaceFolder}/cli/cmd/main.go",
+			"args": ["extract"],
+			"envFile": "${workspaceFolder}/.env"
+		},
+		{
+			"name": "CLI: Transfer",
+			"cwd": "${workspaceFolder}",
+			"type": "go",
+			"request": "launch",
+			"program": "${workspaceFolder}/cli/cmd/main.go",
+			"args": ["transfer"],
+			"envFile": "${workspaceFolder}/.env"
+		},
+		{
+			"name": "CLI: Load",
+			"cwd": "${workspaceFolder}",
+			"type": "go",
+			"request": "launch",
+			"program": "${workspaceFolder}/cli/cmd/main.go",
+			"args": ["load"],
+			"envFile": "${workspaceFolder}/.env"
 		}
 	]
 }

cli/internal/cmd/load.go

@@ -37,6 +37,11 @@ func LoadCmd(ctx context.Context) *cobra.Command {
 					return err
 				}
 
+				err = domain.LoadAccountBalances(ctx, queries, importLogId)
+				if err != nil {
+					return err
+				}
+
 				return nil
 			})
 			if err != nil {

cli/internal/db/query.sql

@@ -119,5 +119,12 @@ INSERT INTO "AccountBalance"("current",
                              "available",
                              "isoCurrencyCode",
                              "accountPlaidId",
-                             "createdAt")
-VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP);
+                             "date",
+                             "importLogId",
+                             "importedAt")
+VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+ON CONFLICT DO UPDATE SET "current"=excluded."current",
+                          "available"=excluded."available",
+                          "isoCurrencyCode"=excluded."isoCurrencyCode",
+                          "accountPlaidId"=excluded."accountPlaidId",
+                          "date"=excluded."date";

cli/internal/db/schema.sql

@@ -65,26 +65,30 @@ CREATE TABLE IF NOT EXISTS "Account" (
     CONSTRAINT "Account_institutionPlaidId_fkey" FOREIGN KEY ("institutionPlaidId") REFERENCES "Institution" ("plaidId") ON DELETE SET NULL ON UPDATE CASCADE,
     CONSTRAINT "Account_importLogId_fkey" FOREIGN KEY ("importLogId") REFERENCES "ImportLog" ("id") ON DELETE SET NULL ON UPDATE CASCADE
 );
-CREATE TABLE IF NOT EXISTS "AccountBalance" (
-    "id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-    "current" REAL NOT NULL,
-    "available" REAL NOT NULL,
-    "isoCurrencyCode" TEXT NOT NULL,
-    "accountPlaidId" TEXT NOT NULL,
-    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    CONSTRAINT "AccountBalance_accountPlaidId_fkey" FOREIGN KEY ("accountPlaidId") REFERENCES "Account" ("plaidId") ON DELETE RESTRICT ON UPDATE CASCADE
-);
 CREATE TABLE IF NOT EXISTS "Budget" (
     "id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
     "name" TEXT NOT NULL,
     "amount" REAL NOT NULL,
     "range" INTEGER NOT NULL
 );
-CREATE TABLE IF NOT EXISTS "BudgetRule" (
+CREATE TABLE IF NOT EXISTS "BudgetFilter" (
     "id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
     "budgetId" INTEGER NOT NULL,
     "column" TEXT NOT NULL,
     "operator" TEXT NOT NULL DEFAULT 'CONTAINS',
     "value" TEXT NOT NULL,
-    CONSTRAINT "BudgetRule_budgetId_fkey" FOREIGN KEY ("budgetId") REFERENCES "Budget" ("id") ON DELETE RESTRICT ON UPDATE CASCADE
+    CONSTRAINT "BudgetFilter_budgetId_fkey" FOREIGN KEY ("budgetId") REFERENCES "Budget" ("id") ON DELETE RESTRICT ON UPDATE CASCADE
+);
+CREATE TABLE IF NOT EXISTS "AccountBalance" (
+    "current" REAL NOT NULL,
+    "available" REAL NOT NULL,
+    "isoCurrencyCode" TEXT NOT NULL,
+    "accountPlaidId" TEXT NOT NULL,
+    "date" TEXT NOT NULL,
+    "importedAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "importLogId" INTEGER,
+
+    PRIMARY KEY ("accountPlaidId", "date"),
+    CONSTRAINT "AccountBalance_accountPlaidId_fkey" FOREIGN KEY ("accountPlaidId") REFERENCES "Account" ("plaidId") ON DELETE RESTRICT ON UPDATE CASCADE,
+    CONSTRAINT "AccountBalance_importLogId_fkey" FOREIGN KEY ("importLogId") REFERENCES "ImportLog" ("id") ON DELETE SET NULL ON UPDATE CASCADE
 );

cli/internal/domain/csv.go

@@ -81,3 +81,10 @@ type Account struct {
 	Type             plaid.AccountType    `csv:"type"`
 	PlaidItemID      string               `csv:"plaidItemId"`
 }
+
+type AccountBalance struct {
+	PlaidAccountID string  `csv:"plaidAccountId"`
+	Available      float64 `csv:"available"`
+	Current        float64 `csv:"current"`
+	Date           string  `csv:"date"`
+}

cli/internal/domain/load.go

@@ -175,18 +175,36 @@ func LoadAccounts(ctx context.Context, queries *db.Queries, importLogId int64) e
 		}
 
 		log.Println("created account", account.Name)
+	}
+
+	return nil
+}
+
+func LoadAccountBalances(ctx context.Context, queries *db.Queries, importLogId int64) error {
+	csvFile := path.Join(os.Getenv("HOME"), ".config", "budgeted", "csv", "account_balances.csv")
+	data, err := os.Open(csvFile)
+	if err != nil {
+		return errors.Wrapf(err, "failed to read file: %s", csvFile)
+	}
+
+	var accountBalances []AccountBalance
+	if err := gocsv.Unmarshal(data, &accountBalances); err != nil {
+		return errors.Wrapf(err, "failed to parse CSV file: %s", csvFile)
+	}
 
+	for _, accountBalance := range accountBalances {
 		err = queries.AccountBalanceCreate(ctx, db.AccountBalanceCreateParams{
-			Current:         account.CurrentBalance,
-			Available:       account.AvailableBalance,
-			IsoCurrencyCode: account.ISOCurrencyCode,
-			AccountPlaidId:  account.PlaidID,
+			AccountPlaidId: accountBalance.PlaidAccountID,
+			Current:        accountBalance.Current,
+			Available:      accountBalance.Available,
+			Date:           accountBalance.Date,
+			ImportLogId:    sql.NullInt64{Int64: importLogId, Valid: true},
 		})
 		if err != nil {
 			return err
 		}
 
-		log.Println("created account balance", account.Name)
+		log.Println("created account balance", accountBalance.Current)
 	}
 
 	return nil

cli/internal/domain/transform.go

@@ -124,23 +124,35 @@ func TransformTransactions(ctx context.Context, jsonStorage string, csvStorage s
 	return nil
 }
 
+// TransformAccounts transforms the JSON account files into CSV files.
+// For accounts, we find the latest JSON file in the instititution's accounts directory.
+// Each ETL run will write a new JSON file with the most up-to-date accounts and account info,
+// so we only need to use the latest file.
+// The unfortunate side-effect of that is rebuilding a SQLite database will no longer rebuild
+// historical account balances.
 func TransformAccounts(ctx context.Context, jsonStorage string, csvStorage string) error {
-	var accountsCSV []Account
 	accountsPath := fmt.Sprintf("%s/accounts", jsonStorage)
+	accountMap := make(map[string]Account)
+	accountBalanceMap := make(map[string]AccountBalance)
+	var accounts []Account
+	var accountBalances []AccountBalance
 
 	err := filepath.WalkDir(accountsPath, func(path string, d os.DirEntry, err error) error {
 		if err != nil {
 			return err
 		}
-
 		if d.IsDir() {
 			return nil
 		}
 
+		log.Println("path", path)
+		date := d.Name()[:10]
+
 		bytes, err := os.ReadFile(path)
 		if err != nil {
-			return err
+			log.Fatal(err)
 		}
+
 		response := plaid.AccountsGetResponse{}
 		if err = json.Unmarshal(bytes, &response); err != nil {
 			return err
@@ -150,7 +162,7 @@ func TransformAccounts(ctx context.Context, jsonStorage string, csvStorage strin
 			balance := account.GetBalances()
 			item := response.GetItem()
 
-			accountsCSV = append(accountsCSV, Account{
+			accountMap[account.GetAccountId()] = Account{
 				PlaidID:          account.GetAccountId(),
 				AvailableBalance: balance.GetAvailable(),
 				CurrentBalance:   balance.GetCurrent(),
@@ -162,22 +174,46 @@ func TransformAccounts(ctx context.Context, jsonStorage string, csvStorage strin
 				Subtype:          account.GetSubtype(),
 				Type:             account.GetType(),
 				PlaidItemID:      item.GetItemId(),
-			})
+			}
+
+			balanceKey := fmt.Sprintf("%s-%s", date, account.GetAccountId())
+
+			accountBalanceMap[balanceKey] = AccountBalance{
+				PlaidAccountID: account.GetAccountId(),
+				Date:           date,
+				Available:      balance.GetAvailable(),
+				Current:        balance.GetCurrent(),
+			}
 		}
 
 		return nil
 	})
+
+	for _, account := range accountMap {
+		accounts = append(accounts, account)
+	}
+	for _, accountBalance := range accountBalanceMap {
+		accountBalances = append(accountBalances, accountBalance)
+	}
+
+	var csvContent string
+	var fp string
+
+	csvContent, err = gocsv.MarshalString(&accounts)
 	if err != nil {
 		return err
 	}
+	fp = filepath.Join(csvStorage, "accounts.csv")
+	if err := os.WriteFile(fp, []byte(csvContent), 0644); err != nil {
+		return err
+	}
 
-	csvContent, err := gocsv.MarshalString(&accountsCSV)
+	csvContent, err = gocsv.MarshalString(&accountBalances)
 	if err != nil {
 		return err
 	}
-
-	filePath := filepath.Join(csvStorage, "accounts.csv")
-	if err := os.WriteFile(filePath, []byte(csvContent), 0644); err != nil {
+	fp = filepath.Join(csvStorage, "account_balances.csv")
+	if err := os.WriteFile(fp, []byte(csvContent), 0644); err != nil {
 		return err
 	}
 

electron/main/api.ts

@@ -248,7 +248,7 @@ export const router = t.router({
 	}),
 	accountBalances: loggedProcedure.query(async () => {
 		return await prisma.accountBalance.findMany({
-			orderBy: { createdAt: 'asc' },
+			orderBy: { date: 'asc' },
 			include: { account: true },
 		})
 	}),

prisma/migrations/20240207000008_uniquify_account_balance_history/migration.sql

@@ -0,0 +1,26 @@
+/*
+  Warnings:
+
+  - The primary key for the `AccountBalance` table will be changed. If it partially fails, the table could be left without primary key constraint.
+  - You are about to drop the column `createdAt` on the `AccountBalance` table. All the data in the column will be lost.
+  - You are about to drop the column `id` on the `AccountBalance` table. All the data in the column will be lost.
+  - Added the required column `date` to the `AccountBalance` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- RedefineTables
+PRAGMA foreign_keys=OFF;
+CREATE TABLE "new_AccountBalance" (
+    "current" REAL NOT NULL,
+    "available" REAL NOT NULL,
+    "isoCurrencyCode" TEXT NOT NULL,
+    "accountPlaidId" TEXT NOT NULL,
+    "date" TEXT NOT NULL,
+
+    PRIMARY KEY ("accountPlaidId", "date"),
+    CONSTRAINT "AccountBalance_accountPlaidId_fkey" FOREIGN KEY ("accountPlaidId") REFERENCES "Account" ("plaidId") ON DELETE RESTRICT ON UPDATE CASCADE
+);
+INSERT INTO "new_AccountBalance" ("accountPlaidId", "available", "current", "isoCurrencyCode") SELECT "accountPlaidId", "available", "current", "isoCurrencyCode" FROM "AccountBalance";
+DROP TABLE "AccountBalance";
+ALTER TABLE "new_AccountBalance" RENAME TO "AccountBalance";
+PRAGMA foreign_key_check;
+PRAGMA foreign_keys=ON;