diff --git a/finnow-api/source/scheduled_jobs.d b/finnow-api/source/scheduled_jobs.d index cac7741..028dcb1 100644 --- a/finnow-api/source/scheduled_jobs.d +++ b/finnow-api/source/scheduled_jobs.d @@ -45,25 +45,18 @@ void startScheduledJobs() { import profile.data; import profile.data_impl_sqlite; import profile.model; - import transaction.dto; - import transaction.data; - import util.pagination; - import std.stdio; - import std.datetime; - import cronexp; + import transaction.service; + import std.datetime.stopwatch; + StopWatch sw = StopWatch(AutoStart.yes); FileSystemProfileRepository.doForAllUserProfiles((Profile profile, ProfileRepository profileRepo) { - writefln!"Recurring transaction check: %s / %s"(profile.username, profile.name); ProfileDataSource ds = profileRepo.getDataSource(profile); - RecurringTransactionRepository rtRepo = ds.getRecurringTransactionRepository(); - Page!(RecurringTransactionResponse) result = rtRepo.findAll(PageRequest.unpaged()); - foreach (rt; result.items) { - writeln(rt.scheduleExpr); - DateTime now = cast(DateTime) Clock.currTime(); - auto cron = CronExpr(rt.scheduleExpr); - writefln!"scheduleExpr = %s -> %s"(rt.scheduleExpr, cron.getNext(now)); - // TODO: Figure out how to actually create the transactions! - } + updateRecurringTransactionScheduling(ds); }); + sw.stop(); + long elapsedTimeMs = sw.peek.total!"msecs"; + if (elapsedTimeMs > 1000) { + warnF!"Took more than 1 second to process recurring transactions: %d ms."(elapsedTimeMs); + } }, new FixedIntervalSchedule(minutes(1), Clock.currTime(UTC()))); jobScheduler.start(); diff --git a/finnow-api/source/transaction/data.d b/finnow-api/source/transaction/data.d index 4466257..5dbba7b 100644 --- a/finnow-api/source/transaction/data.d +++ b/finnow-api/source/transaction/data.d @@ -74,4 +74,9 @@ interface RecurringTransactionRepository { Optional!RecurringTransactionResponse findById(ulong id); RecurringTransactionResponse insert(in RecurringTransactionPayload data); void deleteById(ulong id); + void deleteAllByDraftId(ulong draftId); + + void addToQueue(ulong id, string timestamp); + Optional!string findInQueue(ulong id); + void removeFromQueue(ulong id); } diff --git a/finnow-api/source/transaction/data_impl_sqlite.d b/finnow-api/source/transaction/data_impl_sqlite.d index 2476caf..3243ed2 100644 --- a/finnow-api/source/transaction/data_impl_sqlite.d +++ b/finnow-api/source/transaction/data_impl_sqlite.d @@ -770,6 +770,11 @@ class SqliteTransactionDraftRepository : TransactionDraftRepository { draftId ); insertLineItems(draftId, data); + // If template name is empty / NULL, clear any recurring transactions that may exist. + if (data.templateName.isNull || data.templateName.value.length == 0) { + new SqliteRecurringTransactionRepository(this.db) + .deleteAllByDraftId(draftId); + } return findById(draftId).orElseThrow(); } @@ -957,6 +962,40 @@ class SqliteRecurringTransactionRepository : RecurringTransactionRepository { util.sqlite.deleteById(db, "recurring_transaction", id); } + void deleteAllByDraftId(ulong draftId) { + util.sqlite.update( + db, + "DELETE FROM recurring_transaction WHERE draft_id = ?", + draftId + ); + } + + void addToQueue(ulong id, string timestamp) { + util.sqlite.update( + db, + "INSERT INTO recurring_transaction_queue (recurring_transaction_id, scheduled_timestamp) VALUES (?, ?)", + id, + timestamp + ); + } + + Optional!string findInQueue(ulong id) { + return util.sqlite.findOne( + db, + "SELECT scheduled_timestamp FROM recurring_transaction_queue WHERE recurring_transaction_id = ?", + row => row.peek!string(0), + id + ); + } + + void removeFromQueue(ulong id) { + util.sqlite.update( + db, + "DELETE FROM recurring_transaction_queue WHERE recurring_transaction_id = ?", + id + ); + } + private static RecurringTransactionResponse parseRecurringTransaction(Row row) { return RecurringTransactionResponse( row.peek!ulong(0), diff --git a/finnow-api/source/transaction/service.d b/finnow-api/source/transaction/service.d index 9c6ebf2..9b704ae 100644 --- a/finnow-api/source/transaction/service.d +++ b/finnow-api/source/transaction/service.d @@ -551,3 +551,104 @@ RecurringTransactionResponse[] getRecurringTransactionsForDraft(ProfileDataSourc return ds.getRecurringTransactionRepository() .findAllByDraftId(draftId); } + +/** + * A routine used by a scheduled job that runs periodically for all users, and + * all those users' profiles. It iterates over every one of the user's recurring + * transactions and does the following: + * - Adds it to the persistent queue with the next timestamp if it's not + * already there. + * - Creates a new draft transaction from the recurring transaction's linked + * template, if its queued timestamp has elapsed. Then requeues that + * recurring transaction for its next timestamp. + * Params: + * ds = The profile data source to operate on. + */ +void updateRecurringTransactionScheduling(ProfileDataSource ds) { + import cronexp; + + RecurringTransactionRepository rtRepo = ds.getRecurringTransactionRepository(); + Page!(RecurringTransactionResponse) result = rtRepo.findAll(PageRequest.unpaged()); + const DateTime NOW = cast(DateTime) Clock.currTime(UTC()); + ds.doTransaction(() { + foreach (RecurringTransactionResponse rt; result.items) { + auto cron = CronExpr(rt.scheduleExpr); + auto nextScheduledOccurrence = cron.getNext(NOW); + if (nextScheduledOccurrence.isNull) { + rtRepo.deleteById(rt.id); + } else { + // First, check if it's already in queue to execute (and execute it if it's time). + Optional!string queuedTimestampStr = rtRepo.findInQueue(rt.id); + bool shouldRequeue = false; + if (queuedTimestampStr) { + DateTime queuedTimestamp = DateTime.fromISOExtString(queuedTimestampStr.value); + if (queuedTimestamp < NOW) { + createDraftForRecurringTransaction(ds, rt, queuedTimestamp); + rtRepo.removeFromQueue(rt.id); + shouldRequeue = true; + } else if (queuedTimestamp != nextScheduledOccurrence) { + // The queued timestamp doesn't match the next scheduled occurrence. + // This shouldn't really happen at all, but if it does, we remove it + // so the proper scheduled timestamp can be added. + warnF!("Found a recurring transaction whose queued timestamp " ~ + "doesn't match the next scheduled occurrence as per its " ~ + "Cron schedule expression. Removing that queued transaction " ~ + "and requeueing it using the Cron expression's next date. " ~ + "Recurring Transaction ID: %d; Cron Expression: %s")(rt.id, rt.scheduleExpr); + rtRepo.removeFromQueue(rt.id); + shouldRequeue = true; + } + } else { + shouldRequeue = true; + } + + if (shouldRequeue) { + const nextTimestamp = nextScheduledOccurrence.get().toISOExtString(); + infoF!"Requeueing recurring transaction %d to occur next at %s."(rt.id, nextTimestamp); + rtRepo.addToQueue(rt.id, nextTimestamp); + } + } + } + }); +} + +private void createDraftForRecurringTransaction( + ProfileDataSource ds, + in RecurringTransactionResponse rt, + DateTime queuedTimestamp +) { + TransactionDraftRepository draftRepo = ds.getTransactionDraftRepository(); + Optional!TransactionDraftResponse draftOpt = draftRepo.findById(rt.draftId); + if (!draftOpt) { + warnF!"Couldn't create draft for recurring transaction %d because template draft %d doesn't exist."( + rt.id, rt.draftId + ); + return; + } + TransactionDraftResponse templateDraft = draftOpt.value; + TransactionDraftPayload payload; + payload.timestamp = queuedTimestamp.toISOExtString().toOptional(); + payload.amount = templateDraft.amount; + payload.currencyCode = templateDraft.currency.mapIfPresent!(c => c.code.idup); + string description = templateDraft.description.orElse("") + ~ "\n- Created automatically as a recurring transaction from template: \"" ~ + templateDraft.templateName.orElse("Unknown") ~ "\"."; + payload.description = description.toOptional; + payload.internalTransfer = templateDraft.internalTransfer; + payload.vendorId = templateDraft.vendor.mapIfPresent!(v => v.id); + payload.categoryId = templateDraft.category.mapIfPresent!(c => c.id); + payload.creditedAccountId = templateDraft.creditedAccount.mapIfPresent!(a => a.id); + payload.debitedAccountId = templateDraft.debitedAccount.mapIfPresent!(a => a.id); + payload.tags = templateDraft.tags; + import std.algorithm : map; + import std.array : array; + payload.lineItems = templateDraft.lineItems.map!(li => TransactionDraftPayload.LineItemPayload( + li.valuePerItem, + li.quantity, + li.description, + li.category.mapIfPresent!(c => c.id) + )).array; + auto createdDraft = draftRepo.insert(payload); + draftRepo.updateTags(createdDraft.id, payload.tags); + infoF!"Inserted draft %d for scheduled recurring transaction %d."(createdDraft.id, rt.id); +} diff --git a/finnow-api/sql/migrations/2.sql b/finnow-api/sql/migrations/2.sql index 38b37b5..5896b37 100644 --- a/finnow-api/sql/migrations/2.sql +++ b/finnow-api/sql/migrations/2.sql @@ -71,4 +71,14 @@ CREATE TABLE recurring_transaction ( CONSTRAINT fk_recurring_transaction_draft FOREIGN KEY (draft_id) REFERENCES transaction_draft(id) ON UPDATE CASCADE ON DELETE CASCADE -); \ No newline at end of file +); + +-- A "queue" table where upcoming recurring transactions are posted until it's +-- time to add them for real. This ensures proper handling of recurring transactions. +CREATE TABLE recurring_transaction_queue ( + recurring_transaction_id INTEGER PRIMARY KEY, + scheduled_timestamp TEXT NOT NULL, + CONSTRAINT fk_recurring_transaction_queue_id + FOREIGN KEY (recurring_transaction_id) REFERENCES recurring_transaction(id) + ON UPDATE CASCADE ON DELETE CASCADE +); diff --git a/finnow-api/sql/schema.sql b/finnow-api/sql/schema.sql index 31a3320..c742d67 100644 --- a/finnow-api/sql/schema.sql +++ b/finnow-api/sql/schema.sql @@ -302,3 +302,13 @@ CREATE TABLE recurring_transaction ( FOREIGN KEY (draft_id) REFERENCES transaction_draft(id) ON UPDATE CASCADE ON DELETE CASCADE ); + +-- A "queue" table where upcoming recurring transactions are posted until it's +-- time to add them for real. This ensures proper handling of recurring transactions. +CREATE TABLE recurring_transaction_queue ( + recurring_transaction_id INTEGER PRIMARY KEY, + scheduled_timestamp TEXT NOT NULL, + CONSTRAINT fk_recurring_transaction_queue_id + FOREIGN KEY (recurring_transaction_id) REFERENCES recurring_transaction(id) + ON UPDATE CASCADE ON DELETE CASCADE +); diff --git a/web-app/package.json b/web-app/package.json index a42b96f..30c5ada 100644 --- a/web-app/package.json +++ b/web-app/package.json @@ -26,6 +26,7 @@ "chart.js": "^4.5.1", "chartjs-adapter-date-fns": "^3.0.0", "cron-parser": "^5.6.1", + "cronstrue": "^3.24.0", "date-fns": "^4.1.0", "date-fns-tz": "^3.2.0", "pinia": "^3.0.4", diff --git a/web-app/src/components/AddRecurringTransactionModal.vue b/web-app/src/components/AddRecurringTransactionModal.vue index 7a01c3f..9ca784b 100644 --- a/web-app/src/components/AddRecurringTransactionModal.vue +++ b/web-app/src/components/AddRecurringTransactionModal.vue @@ -5,13 +5,14 @@ import { type RecurringTransactionResponse, } from '@/api/transaction' import ModalWrapper from './common/ModalWrapper.vue' -import { ref, useTemplateRef, type Ref } from 'vue' +import { computed, ref, useTemplateRef, watch, type Ref } from 'vue' import { useRoute } from 'vue-router' import AppForm from './common/form/AppForm.vue' import FormGroup from './common/form/FormGroup.vue' import FormControl from './common/form/FormControl.vue' import AppButton from './common/AppButton.vue' import { getSelectedProfile } from '@/api/profile.ts' +import { CronExpressionParser } from 'cron-parser' const route = useRoute() const props = defineProps<{ draftId: number }>() @@ -19,7 +20,34 @@ const modal = useTemplateRef('modal') const savedTxn: Ref = ref() // Form Data: +enum ScheduleType { + Monthly, + Custom, +} + +const chosenScheduleType: Ref = ref(ScheduleType.Monthly) +const inputIsValid = ref(true) const dayOfMonth: Ref = ref(1) +const customScheduleText = ref('') +const customCronErrorMsg = ref(null) + +// Reset form data when the user chooses a different schedule type. +watch(chosenScheduleType, () => { + dayOfMonth.value = 1 + customScheduleText.value = '' + customCronErrorMsg.value = null +}) + +watch([chosenScheduleType, dayOfMonth, customScheduleText], () => { + inputIsValid.value = checkInputValid() +}) + +const upcomingDates = computed(() => { + if (!inputIsValid.value) return [] + const cronExpr = getScheduleExpression() + const cron = CronExpressionParser.parse(cronExpr) + return cron.take(3).map((d) => d.toString()) +}) async function show(): Promise { if (!modal.value) return undefined @@ -34,7 +62,7 @@ async function show(): Promise { async function addRecurringTransaction() { const payload: RecurringTransactionPayload = { draftId: props.draftId, - scheduleExpr: `0 0 0 ${dayOfMonth.value} * *`, + scheduleExpr: getScheduleExpression(), } const api = new TransactionApiClient(getSelectedProfile(route)) try { @@ -46,6 +74,48 @@ async function addRecurringTransaction() { } } +function getScheduleExpression(): string { + if (chosenScheduleType.value === ScheduleType.Custom) { + try { + CronExpressionParser.parse(customScheduleText.value) + return customScheduleText.value.trim() + } catch (err) { + console.warn('Failed to parse schedule Cron expression.', err) + throw err + } + } else if (chosenScheduleType.value === ScheduleType.Monthly) { + return `0 0 12 ${dayOfMonth.value} * *` + } + throw new Error('Unsupported schedule type.') +} + +function checkInputValid(): boolean { + if (chosenScheduleType.value === ScheduleType.Custom) { + customCronErrorMsg.value = null + if (customScheduleText.value.trim().length === 0) { + customCronErrorMsg.value = 'Missing cron expression.' + return false + } + try { + const expr = CronExpressionParser.parse(customScheduleText.value) + if (expr.hasNext()) { + return true + } else { + customCronErrorMsg.value = 'This cron expression has no discernable next date.' + return false + } + } catch (err) { + if (err && typeof err === 'object' && 'message' in err && typeof err.message === 'string') { + customCronErrorMsg.value = err.message + } + return false + } + } else if (chosenScheduleType.value === ScheduleType.Monthly) { + return dayOfMonth.value >= 1 && dayOfMonth.value <= 28 + } + return false +} + defineExpose({ show })