1use crate::ports::{Database, DbError, Statement};
21use std::collections::HashMap;
22use std::time::Duration as StdDuration;
23use time::{Date, Month, OffsetDateTime, UtcOffset};
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub struct Months(pub u32);
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum Period {
34 CalendarMonthUtc,
36 Anchored {
44 anchor: OffsetDateTime,
45 every: Months,
46 },
47 Day,
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub struct PeriodWindow {
54 pub start: OffsetDateTime,
55 pub end: OffsetDateTime,
56}
57
58impl Period {
59 #[must_use]
61 pub fn window_at(self, now: OffsetDateTime) -> PeriodWindow {
62 self.window_for(self.index_at(now))
63 }
64
65 #[must_use]
68 pub fn index_at(self, now: OffsetDateTime) -> i64 {
69 let now = now.to_offset(UtcOffset::UTC);
70 match self {
71 Self::CalendarMonthUtc => month_index(now.year(), now.month()),
72 Self::Day => i64::from(now.date().to_julian_day()),
73 Self::Anchored { anchor, every } => {
74 let anchor = anchor.to_offset(UtcOffset::UTC);
75 let every = i64::from(every.0.max(1));
76 let diff = month_index(now.year(), now.month())
79 - month_index(anchor.year(), anchor.month());
80 let mut index = diff.div_euclid(every);
81 while anchored_start(anchor, every, index + 1) <= now {
82 index += 1;
83 }
84 while anchored_start(anchor, every, index) > now {
85 index -= 1;
86 }
87 index
88 }
89 }
90 }
91
92 #[must_use]
94 pub fn window_for(self, index: i64) -> PeriodWindow {
95 match self {
96 Self::CalendarMonthUtc => PeriodWindow {
97 start: month_start(index),
98 end: month_start(index + 1),
99 },
100 Self::Day => PeriodWindow {
101 start: day_start(index),
102 end: day_start(index + 1),
103 },
104 Self::Anchored { anchor, every } => {
105 let every = i64::from(every.0.max(1));
106 PeriodWindow {
107 start: anchored_start(anchor, every, index),
108 end: anchored_start(anchor, every, index + 1),
109 }
110 }
111 }
112 }
113}
114
115fn month_index(year: i32, month: Month) -> i64 {
118 i64::from(year) * 12 + i64::from(u8::from(month)) - 1
119}
120
121fn month_start(index: i64) -> OffsetDateTime {
123 let year = i32::try_from(index.div_euclid(12)).expect("month index within time's year range");
124 let month = month_from(index.rem_euclid(12) + 1);
125 Date::from_calendar_date(year, month, 1)
126 .expect("the first of a valid month is a valid date")
127 .midnight()
128 .assume_utc()
129}
130
131fn day_start(index: i64) -> OffsetDateTime {
134 Date::from_julian_day(i32::try_from(index).expect("julian day within time's year range"))
135 .expect("a valid julian day is a valid date")
136 .midnight()
137 .assume_utc()
138}
139
140fn anchored_start(anchor: OffsetDateTime, every: i64, index: i64) -> OffsetDateTime {
144 let anchor = anchor.to_offset(UtcOffset::UTC);
145 let target = month_index(anchor.year(), anchor.month()) + index * every;
146 let year = i32::try_from(target.div_euclid(12)).expect("anchor month within time's year range");
147 let month = month_from(target.rem_euclid(12) + 1);
148 let day = anchor.day().min(days_in_month(year, month));
149 Date::from_calendar_date(year, month, day)
150 .expect("the clamped day is valid for its month")
151 .with_hms(anchor.hour(), anchor.minute(), anchor.second())
152 .expect("the anchor's time-of-day is valid")
153 .assume_utc()
154}
155
156fn month_from(one_based: i64) -> Month {
158 let number = u8::try_from(one_based).expect("1..=12 fits u8");
159 Month::try_from(number).expect("1..=12 is a month")
160}
161
162fn days_in_month(year: i32, month: Month) -> u8 {
165 match month {
166 Month::February if is_leap_year(year) => 29,
167 Month::February => 28,
168 Month::April | Month::June | Month::September | Month::November => 30,
169 _ => 31,
170 }
171}
172
173fn is_leap_year(year: i32) -> bool {
175 (year % 4 == 0 && year % 100 != 0) || year % 400 == 0
176}
177
178pub(crate) fn rfc3339(at: OffsetDateTime) -> String {
182 use time::format_description::well_known::Rfc3339;
183 let at = at.to_offset(UtcOffset::UTC);
184 at.replace_nanosecond(0)
185 .unwrap_or(at)
186 .format(&Rfc3339)
187 .unwrap_or_default()
188}
189
190fn counter(value: u64) -> i64 {
194 i64::try_from(value).unwrap_or(i64::MAX)
195}
196
197fn retry_after(period_end: OffsetDateTime, now: OffsetDateTime) -> StdDuration {
200 let millis = (period_end - now).whole_milliseconds().max(0);
201 StdDuration::from_millis(u64::try_from(millis).unwrap_or(u64::MAX))
202}
203
204#[derive(Debug, Clone, Copy, PartialEq, Eq)]
206pub struct Consumed {
207 pub used: u64,
209 pub limit: Option<u64>,
211 pub period_end: OffsetDateTime,
213}
214
215#[derive(Debug, Clone, Copy, PartialEq, Eq)]
220pub struct Exhausted {
221 pub used: u64,
223 pub limit: u64,
225 pub period_end: OffsetDateTime,
227 pub retry_after: StdDuration,
230}
231
232impl Exhausted {
233 #[must_use]
238 pub fn spent(used: u64, limit: u64, period: Period, now: OffsetDateTime) -> Self {
239 let end = period.window_at(now).end;
240 Self {
241 used,
242 limit,
243 period_end: end,
244 retry_after: retry_after(end, now),
245 }
246 }
247}
248
249#[derive(Debug, Clone, Copy, PartialEq, Eq)]
251pub enum Consumption {
252 Consumed(Consumed),
253 Exhausted(Exhausted),
254}
255
256#[derive(Debug, Clone)]
263pub struct Usage {
264 table: String,
265}
266
267impl Usage {
268 #[must_use]
269 pub fn new(table: impl Into<String>) -> Self {
270 Self {
271 table: table.into(),
272 }
273 }
274
275 #[must_use]
278 pub fn create_table_sql(&self) -> String {
279 format!(
280 "CREATE TABLE IF NOT EXISTS {table} (\n \
281 subject TEXT NOT NULL,\n \
282 meter TEXT NOT NULL,\n \
283 period_start TEXT NOT NULL,\n \
284 used BIGINT NOT NULL,\n \
285 PRIMARY KEY (subject, meter, period_start)\n);",
286 table = self.table
287 )
288 }
289
290 #[must_use]
296 pub fn consume_statement(
297 &self,
298 subject: &str,
299 meter: &str,
300 period: Period,
301 now: OffsetDateTime,
302 amount: u64,
303 limit: Option<u64>,
304 ) -> Statement {
305 let start = rfc3339(period.window_at(now).start);
306 match limit {
307 Some(limit) => Statement::with_values(
308 guard_sql(&self.table),
309 vec![
310 subject.to_owned().into(),
311 meter.to_owned().into(),
312 start.into(),
313 counter(amount).into(),
314 counter(limit).into(),
315 counter(amount).into(),
316 counter(limit).into(),
317 ],
318 ),
319 None => Statement::with_values(
320 plain_sql(&self.table),
321 vec![
322 subject.to_owned().into(),
323 meter.to_owned().into(),
324 start.into(),
325 counter(amount).into(),
326 ],
327 ),
328 }
329 }
330
331 #[allow(clippy::too_many_arguments)]
345 pub async fn consume(
346 &self,
347 db: &dyn Database,
348 subject: &str,
349 meter: &str,
350 period: Period,
351 now: OffsetDateTime,
352 amount: u64,
353 limit: Option<u64>,
354 ) -> Result<Consumption, DbError> {
355 let window = period.window_at(now);
356 let start = rfc3339(window.start);
357 let (sql, values) = match limit {
361 Some(limit) => (
362 guarded_sql(&self.table),
363 vec![
364 subject.to_owned().into(),
365 meter.to_owned().into(),
366 start.into(),
367 counter(amount).into(),
368 counter(amount).into(),
369 counter(limit).into(),
370 counter(limit).into(),
371 ],
372 ),
373 None => (
374 plain_sql(&self.table),
375 vec![
376 subject.to_owned().into(),
377 meter.to_owned().into(),
378 start.into(),
379 counter(amount).into(),
380 ],
381 ),
382 };
383 let affected = db.execute(&Statement::with_values(sql, values)).await?;
384 let used = self.read(db, subject, meter, period, now).await?;
385 Ok(match limit {
386 Some(limit) if affected == 0 => Consumption::Exhausted(Exhausted {
387 used,
388 limit,
389 period_end: window.end,
390 retry_after: retry_after(window.end, now),
391 }),
392 _ => Consumption::Consumed(Consumed {
393 used,
394 limit,
395 period_end: window.end,
396 }),
397 })
398 }
399
400 pub async fn refund(
407 &self,
408 db: &dyn Database,
409 subject: &str,
410 meter: &str,
411 period: Period,
412 now: OffsetDateTime,
413 amount: u64,
414 ) -> Result<(), DbError> {
415 let start = rfc3339(period.window_at(now).start);
416 let stmt = Statement::with_values(
419 format!(
420 "UPDATE {table} SET used = CASE WHEN used > ? THEN used - ? ELSE 0 END\n\
421 WHERE subject = ? AND meter = ? AND period_start = ?",
422 table = self.table
423 ),
424 vec![
425 counter(amount).into(),
426 counter(amount).into(),
427 subject.to_owned().into(),
428 meter.to_owned().into(),
429 start.into(),
430 ],
431 );
432 db.execute(&stmt).await?;
433 Ok(())
434 }
435
436 pub async fn read(
443 &self,
444 db: &dyn Database,
445 subject: &str,
446 meter: &str,
447 period: Period,
448 now: OffsetDateTime,
449 ) -> Result<u64, DbError> {
450 let start = rfc3339(period.window_at(now).start);
451 let rows = db
452 .query(&Statement::with_values(
453 format!(
454 "SELECT used FROM {table} WHERE subject = ? AND meter = ? AND period_start = ?",
455 table = self.table
456 ),
457 vec![
458 subject.to_owned().into(),
459 meter.to_owned().into(),
460 start.into(),
461 ],
462 ))
463 .await?;
464 Ok(rows
465 .first()
466 .and_then(|row| row.get::<u64>("used"))
467 .unwrap_or(0))
468 }
469
470 pub async fn history(
478 &self,
479 db: &dyn Database,
480 subject: &str,
481 meter: &str,
482 period: Period,
483 now: OffsetDateTime,
484 last_n_periods: u32,
485 ) -> Result<Vec<(PeriodWindow, u64)>, DbError> {
486 let n = i64::from(last_n_periods);
487 if n == 0 {
488 return Ok(Vec::new());
489 }
490 let current = period.index_at(now);
491 let oldest = period.window_for(current - n + 1).start;
492 let newest_end = period.window_for(current).end;
493 let rows = db
494 .query(&Statement::with_values(
495 format!(
496 "SELECT period_start, used FROM {table} \
497 WHERE subject = ? AND meter = ? AND period_start >= ? AND period_start < ?",
498 table = self.table
499 ),
500 vec![
501 subject.to_owned().into(),
502 meter.to_owned().into(),
503 rfc3339(oldest).into(),
504 rfc3339(newest_end).into(),
505 ],
506 ))
507 .await?;
508 let mut spent: HashMap<String, u64> = HashMap::new();
509 for row in &rows.rows {
510 if let (Some(start), Some(used)) =
511 (row.get::<String>("period_start"), row.get::<u64>("used"))
512 {
513 spent.insert(start, used);
514 }
515 }
516 Ok((0..n)
517 .map(|back| {
518 let window = period.window_for(current - back);
519 let used = spent.get(&rfc3339(window.start)).copied().unwrap_or(0);
520 (window, used)
521 })
522 .collect())
523 }
524
525 pub async fn purge(
538 &self,
539 db: &dyn Database,
540 period: Period,
541 now: OffsetDateTime,
542 keep_periods: u32,
543 ) -> Result<u64, DbError> {
544 let oldest_kept = period.index_at(now) - i64::from(keep_periods) + 1;
545 let cutoff = rfc3339(period.window_for(oldest_kept).start);
546 db.execute(&Statement::with_values(
547 format!(
548 "DELETE FROM {table} WHERE period_start < ?",
549 table = self.table
550 ),
551 vec![cutoff.into()],
552 ))
553 .await
554 }
555}
556
557fn guard_sql(table: &str) -> String {
561 format!(
562 "INSERT INTO {table} (subject, meter, period_start, used)\n\
563 VALUES (?, ?, ?, CASE WHEN ? <= ? THEN ? ELSE NULL END)\n\
564 ON CONFLICT (subject, meter, period_start) DO UPDATE\n\
565 SET used = CASE WHEN {table}.used + excluded.used <= ? \
566 THEN {table}.used + excluded.used ELSE NULL END"
567 )
568}
569
570fn plain_sql(table: &str) -> String {
572 format!(
573 "INSERT INTO {table} (subject, meter, period_start, used)\n\
574 VALUES (?, ?, ?, ?)\n\
575 ON CONFLICT (subject, meter, period_start) DO UPDATE\n\
576 SET used = {table}.used + excluded.used"
577 )
578}
579
580fn guarded_sql(table: &str) -> String {
585 format!(
586 "INSERT INTO {table} (subject, meter, period_start, used)\n\
587 SELECT ?, ?, ?, ? WHERE ? <= ?\n\
588 ON CONFLICT (subject, meter, period_start) DO UPDATE\n\
589 SET used = {table}.used + excluded.used\n\
590 WHERE {table}.used + excluded.used <= ?"
591 )
592}
593
594#[cfg(test)]
595mod tests {
596 use super::*;
597 use crate::lint_portable_sql;
598 use time::format_description::well_known::Rfc3339;
599
600 fn at(text: &str) -> OffsetDateTime {
603 OffsetDateTime::parse(text, &Rfc3339).expect("test instant parses")
604 }
605
606 fn window(start: &str, end: &str) -> PeriodWindow {
607 PeriodWindow {
608 start: at(start),
609 end: at(end),
610 }
611 }
612
613 #[test]
614 fn create_table_sql_is_portable_ddl() {
615 let sql = Usage::new("waitlist_usage").create_table_sql();
616 assert!(sql.contains("used BIGINT NOT NULL"));
617 assert_eq!(lint_portable_sql(&sql), vec![], "DDL must be portable");
618 }
619
620 #[test]
621 fn the_statement_shapes_bind_seven_or_four_values() {
622 let usage = Usage::new("waitlist_usage");
623 let period = Period::CalendarMonthUtc;
624 let now = at("2026-01-15T00:00:00Z");
625
626 let guarded = usage.consume_statement("a", "mail", period, now, 1, Some(100));
627 assert!(guarded.sql.contains("ELSE NULL END"));
628 assert!(guarded.sql.contains("excluded.used"));
629 assert_eq!(guarded.values.0.len(), 7);
631
632 let unbounded = usage.consume_statement("a", "mail", period, now, 1, None);
633 assert_eq!(unbounded.values.0.len(), 4);
634
635 let huge = usage.consume_statement("a", "mail", period, now, u64::MAX, Some(u64::MAX));
637 assert!(
638 matches!(huge.values.0[3], sea_query::Value::BigInt(Some(v)) if v == i64::MAX),
639 "amount must saturate to i64::MAX"
640 );
641 }
642
643 #[test]
644 fn calendar_and_day_windows_are_utc() {
645 assert_eq!(
647 Period::CalendarMonthUtc.window_at(at("2026-12-15T00:00:00Z")),
648 window("2026-12-01T00:00:00Z", "2027-01-01T00:00:00Z")
649 );
650 assert_eq!(
652 Period::Day.window_at(at("2026-03-01T23:59:59Z")),
653 window("2026-03-01T00:00:00Z", "2026-03-02T00:00:00Z")
654 );
655 let local = at("2026-03-01T00:30:00+05:30");
658 assert_eq!(
659 Period::Day.window_at(local),
660 window("2026-02-28T00:00:00Z", "2026-03-01T00:00:00Z")
661 );
662 assert_eq!(
663 Period::Day.window_at(local),
664 Period::Day.window_at(local.to_offset(UtcOffset::UTC))
665 );
666 }
667
668 #[test]
669 fn anchored_periods_clamp_the_day_to_the_target_month() {
670 let cases = [
674 (
676 "2026-01-31T09:30:00Z",
677 1,
678 "2026-02-28T10:00:00Z",
679 "2026-02-28T09:30:00Z",
680 "2026-03-31T09:30:00Z",
681 ),
682 (
684 "2026-01-31T09:30:00Z",
685 1,
686 "2026-04-05T00:00:00Z",
687 "2026-03-31T09:30:00Z",
688 "2026-04-30T09:30:00Z",
689 ),
690 (
692 "2026-01-30T09:00:00Z",
693 1,
694 "2026-02-10T00:00:00Z",
695 "2026-01-30T09:00:00Z",
696 "2026-02-28T09:00:00Z",
697 ),
698 (
699 "2026-01-30T09:00:00Z",
700 1,
701 "2026-03-10T00:00:00Z",
702 "2026-02-28T09:00:00Z",
703 "2026-03-30T09:00:00Z",
704 ),
705 (
708 "2024-01-29T00:00:00Z",
709 1,
710 "2024-02-05T00:00:00Z",
711 "2024-01-29T00:00:00Z",
712 "2024-02-29T00:00:00Z",
713 ),
714 (
715 "2025-01-29T00:00:00Z",
716 1,
717 "2025-02-28T00:00:00Z",
718 "2025-02-28T00:00:00Z",
719 "2025-03-29T00:00:00Z",
720 ),
721 (
723 "2026-01-31T00:00:00Z",
724 3,
725 "2026-02-15T00:00:00Z",
726 "2026-01-31T00:00:00Z",
727 "2026-04-30T00:00:00Z",
728 ),
729 (
730 "2026-01-31T00:00:00Z",
731 3,
732 "2026-05-01T00:00:00Z",
733 "2026-04-30T00:00:00Z",
734 "2026-07-31T00:00:00Z",
735 ),
736 ];
737 for (anchor, every, now, start, end) in cases {
738 let period = Period::Anchored {
739 anchor: at(anchor),
740 every: Months(every),
741 };
742 assert_eq!(
743 period.window_at(at(now)),
744 window(start, end),
745 "anchor {anchor}, every {every}, now {now}"
746 );
747 }
748
749 let period = Period::Anchored {
751 anchor: at("2026-03-15T08:00:00Z"),
752 every: Months(1),
753 };
754 assert_eq!(
755 period.window_at(at("2026-01-01T00:00:00Z")),
756 window("2025-12-15T08:00:00Z", "2026-01-15T08:00:00Z")
757 );
758 assert_eq!(period.index_at(at("2026-01-01T00:00:00Z")), -3);
759 }
760
761 #[test]
762 fn index_and_window_are_inverses() {
763 for period in [
764 Period::CalendarMonthUtc,
765 Period::Day,
766 Period::Anchored {
767 anchor: at("2026-01-31T09:30:00Z"),
768 every: Months(1),
769 },
770 ] {
771 let now = at("2026-08-31T09:30:00Z");
772 let index = period.index_at(now);
773 assert_eq!(period.window_for(index), period.window_at(now));
774 assert_eq!(
775 period.window_for(index - 1).end,
776 period.window_for(index).start,
777 "consecutive windows abut"
778 );
779 }
780 }
781}