forcen/accounters
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Moved accumulation calculations into DB triggers

Browse source
This commit is contained in:
Manuel Forcén Muñoz 2024-03-21 23:34:43 +01:00
parent e336292db4
commit 79ef859fbe
9 changed files with 50 additions and 281 deletions
Download patch file Download diff file Expand all files Collapse all files

48
migrations/20231110161954_base.sql
View file

@ -14,14 +14,6 @@ CREATE TABLE IF NOT EXISTS accounts(
FOREIGN KEY (user) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS account_snapshot(
account INTEGER,
datestamp DATE,
amount INT,
FOREIGN KEY (account) REFERENCES accounts(account_id),
PRIMARY KEY (account, datestamp)
);
CREATE TABLE IF NOT EXISTS categories (
category_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT,
@ -41,13 +33,49 @@ CREATE TABLE IF NOT EXISTS transactions (
transaction_id INTEGER PRIMARY KEY AUTOINCREMENT,
account INTEGER,
description TEXT,
transaction_timestamp DATETIME,
tx_date DATETIME,
category INTEGER,
amount INTEGER,
accumulated INTEGER DEFAULT 0,
tx_order INTEGER DEFAULT 0,
hash TEXT,
FOREIGN KEY (account) REFERENCES accounts(account_id),
FOREIGN KEY (category) REFERENCES categories(category_id)
);
CREATE INDEX idx_transactions_ts ON transactions(account, transaction_timestamp);
CREATE TRIGGER tx_insert AFTER INSERT ON transactions
BEGIN
UPDATE transactions
SET accumulated=old.acc+NEW.amount
FROM (
SELECT COALESCE(max(accumulated), 0) AS acc
FROM transactions
WHERE date <= NEW.date
ORDER BY tx_order DESC
LIMIT 1
) AS old
WHERE id=NEW.id;
UPDATE transactions
SET tx_order=old.tx_order+1 FROM (
SELECT COALESCE(max(tx_order), 0) as tx_order
FROM tx WHERE date=NEW.date
) AS old
WHERE id=NEW.id;
UPDATE transactions SET accumulated=calc.acc+NEW.accumulated FROM (
SELECT tx.id, (
SUM(amount) OVER (
ORDER BY date, tx_order
ROWS BETWEEN
UNBOUNDED PRECEDING
AND CURRENT ROW
)
) acc
FROM transactions tx
WHERE date > NEW.date OR id=NEW.id;
)
WHERE transactions.id=calc.id;
END;
CREATE INDEX idx_transactions_ts ON transactions(account, tx_date);
CREATE INDEX idx_transactions_hash ON transactions(hash);

5
src/models.rs
View file

@ -1,8 +1,5 @@
mod account;
pub mod account;
pub mod categories;
pub mod rules;
pub mod transaction;
pub mod users;
pub use account::{Account, AccountSnapshot};
pub use transaction::Transaction;

244
src/models/account.rs
View file

@ -1,199 +1,8 @@
use chrono::{prelude::*, Duration, DurationRound};
use chrono::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::{FromRow, Result, SqlitePool};
use super::{rules::Rule, Transaction};
#[derive(FromRow, Serialize, Deserialize, Clone, Debug)]
pub struct AccountSnapshot {
account: i32,
datestamp: DateTime<Utc>,
amount: i32,
}
impl AccountSnapshot {
pub async fn get(
pool: &SqlitePool,
account: i32,
date: DateTime<Utc>,
) -> Result<AccountSnapshot> {
sqlx::query("SELECT * FROM account_snapshot WHERE account=? AND datestamp=?")
.bind(account)
.bind(date)
.fetch_one(pool)
.await
.and_then(|r| AccountSnapshot::from_row(&r))
}
pub async fn get_last(
pool: &SqlitePool,
account: i32,
date: DateTime<Utc>,
) -> Result<AccountSnapshot> {
sqlx::query("SELECT * FROM account_snapshot WHERE account=? AND datestamp<=? LIMIT 1")
.bind(account)
.bind(date)
.fetch_one(pool)
.await
.and_then(|r| AccountSnapshot::from_row(&r))
}
pub async fn list(
pool: &SqlitePool,
account: i32,
limit: Option<i32>,
offset: Option<i32>,
asc: bool,
) -> sqlx::Result<Vec<AccountSnapshot>> {
let mut query = sqlx::QueryBuilder::new("SELECT * FROM account_snapshot WHERE account=");
query.push_bind(account);
if let Some(limit) = limit {
query.push(" LIMIT ");
query.push_bind(limit);
}
if let Some(offset) = offset {
query.push(" OFFSET ");
query.push_bind(offset);
}
if asc {
query.push(" ORDER BY datestamp ASC");
} else {
query.push(" ORDER BY datestamp DESC");
}
let rows = query.build().fetch_all(pool).await?;
let mut res = Vec::new();
for r in rows.iter() {
res.push(AccountSnapshot::from_row(r)?);
}
Ok(res)
}
pub async fn list_by_date(
pool: &SqlitePool,
account: i32,
after: Option<DateTime<Utc>>,
before: Option<DateTime<Utc>>,
limit: Option<i32>,
asc: bool,
) -> sqlx::Result<Vec<AccountSnapshot>> {
let mut query = sqlx::QueryBuilder::new("SELECT * FROM account_snapshot WHERE account=");
query.push_bind(account);
if let Some(after) = after {
query.push(" AND datestamp >= ");
query.push_bind(after);
}
if let Some(before) = before {
query.push(" AND datestamp < ");
query.push_bind(before);
}
if let Some(limit) = limit {
query.push(" LIMIT ");
query.push_bind(limit);
}
if asc {
query.push(" ORDER BY datestamp ASC");
} else {
query.push(" ORDER BY datestamp DESC");
}
let rows = query.build().fetch_all(pool).await?;
let mut res = Vec::new();
for r in rows.iter() {
res.push(AccountSnapshot::from_row(r)?);
}
Ok(res)
}
pub async fn delete_by_dates(
pool: &SqlitePool,
account: i32,
after: Option<DateTime<Utc>>,
before: Option<DateTime<Utc>>,
) -> sqlx::Result<()> {
if after.is_none() && before.is_none() {
return Err(sqlx::Error::RowNotFound);
}
let mut query = sqlx::QueryBuilder::new("DELETE FROM account_snapshot WHERE account=");
query.push_bind(account);
if let Some(after) = after {
query.push(" AND datestamp >= ");
query.push_bind(after);
}
if let Some(before) = before {
query.push(" AND datestamp < ");
query.push_bind(before);
}
query.build().execute(pool).await?;
Ok(())
}
pub async fn insert(&self, pool: &SqlitePool) -> sqlx::Result<()> {
sqlx::query("INSERT INTO account_snapshot(account, datestamp, amount) VALUES(?,?,?)")
.bind(self.account)
.bind(self.datestamp)
.bind(self.amount)
.execute(pool)
.await
.map(|_| ())
}
pub async fn get_next(&self, pool: &SqlitePool) -> sqlx::Result<Option<AccountSnapshot>> {
let date_next = match Transaction::list_by_date(
pool,
self.account,
Some(self.datestamp + Duration::days(1)),
None,
Some(1),
true,
)
.await?
.first()
{
Some(tx) => tx.get_timestamp(),
None => {
return Ok(None);
}
}
.duration_trunc(chrono::Duration::days(1))
.unwrap();
println!(
"Starting date: {:?}, ending date: {:?}",
self.datestamp, date_next
);
let tx_list = Transaction::list_by_date(
pool,
self.account,
Some(self.datestamp),
Some(date_next),
None,
true,
)
.await?;
Ok(Some(AccountSnapshot {
datestamp: date_next,
account: self.account,
amount: self.amount + tx_list.iter().fold(0, |acc, tx| acc + tx.get_amount()),
}))
}
}
use super::{rules::Rule, transaction::Transaction};
#[derive(FromRow, Serialize, Deserialize, Debug)]
pub struct Account {
@ -254,55 +63,6 @@ impl Account {
Ok(res)
}
pub async fn recalculate_snapshots(
&self,
pool: &SqlitePool,
from: Option<DateTime<Utc>>,
) -> Result<()> {
let mut snap = match from {
Some(f) => {
let snapshot = AccountSnapshot::list_by_date(
pool,
self.get_id(),
None,
Some(f),
Some(1),
true,
)
.await?;
if snapshot.is_empty() {
AccountSnapshot {
account: self.account_id,
datestamp: Utc.timestamp_opt(0, 0).unwrap(),
amount: 0,
}
} else {
snapshot.first().unwrap().clone()
}
}
None => AccountSnapshot {
account: self.account_id,
datestamp: Utc.timestamp_opt(0, 0).unwrap(),
amount: 0,
},
};
AccountSnapshot::delete_by_dates(
pool,
self.get_id(),
Some(snap.datestamp + Duration::hours(12)),
None,
)
.await?;
while let Some(next) = snap.get_next(pool).await? {
next.insert(pool).await?;
snap = next;
}
Ok(())
}
pub async fn recategorize_transactions(
&self,
pool: &SqlitePool,

1
src/models/transaction.rs
View file

@ -20,6 +20,7 @@ pub struct Transaction {
transaction_timestamp: DateTime<Utc>,
category: Option<i32>,
amount: i32,
accumulated: i32,
#[serde(default, skip_serializing)]
hash: Option<String>,
}

4
webserver/src/main.rs
View file

@ -85,10 +85,6 @@ async fn main() {
"/accounts/id/:id/transaction",
get(routes::api::transactions::list),
)
.route(
"/accounts/id/:id/update",
post(routes::api::accounts::snapshot_update),
)
.route(
"/accounts/id/:id/recategorize",
post(routes::api::accounts::recategorize),

18
webserver/src/routes/api/accounts.rs
View file

@ -6,7 +6,7 @@ use serde::Deserialize;
use sqlx::SqlitePool;
use crate::users::UserToken;
use accounters::models::Account;
use accounters::models::account::Account;
pub async fn account_get(
State(db): State<Arc<SqlitePool>>,
@ -51,22 +51,6 @@ pub async fn account_list(
}
}
pub async fn snapshot_update(
State(db): State<Arc<SqlitePool>>,
uid: UserToken,
Path(account): Path<i32>,
) -> (StatusCode, String) {
let account = Account::get_by_id(db.as_ref(), account).await.unwrap();
if account.get_user() != uid.user_id {
return (StatusCode::UNAUTHORIZED, String::new());
}
match account.recalculate_snapshots(db.as_ref(), None).await {
Ok(_) => (StatusCode::OK, String::new()),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")),
}
}
pub async fn recategorize(
State(db): State<Arc<SqlitePool>>,
uid: UserToken,

2
webserver/src/routes/api/transactions.rs
View file

@ -6,7 +6,7 @@ use hyper::StatusCode;
use serde::Deserialize;
use sqlx::SqlitePool;
use accounters::models::{transaction::TxConflictResolutionMode, Transaction};
use accounters::models::transaction::{Transaction, TxConflictResolutionMode};
#[derive(Deserialize)]
pub struct TransactionContent {

2
webserver/src/routes/ui.rs
View file

@ -6,7 +6,7 @@ use sqlx::SqlitePool;
use tera::{Context, Tera};
use crate::users::UserToken;
use accounters::models::{Account, Transaction};
use accounters::models::{account::Account, transaction::Transaction};
pub mod account;
pub mod categories;

5
webserver/src/routes/ui/account.rs
View file

@ -12,7 +12,10 @@ use sqlx::SqlitePool;
use tera::{Context, Tera};
use crate::users::UserToken;
use accounters::models::{transaction::TxConflictResolutionMode, Account, Transaction};
use accounters::models::{
account::Account,
transaction::{Transaction, TxConflictResolutionMode},
};
#[derive(Deserialize)]
pub struct AccountViewParams {