comi-logic/cfw-autumn
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: merge remote-tracking branch 'origin/dev' into agent3

Browse Source
This commit is contained in:
Charlie Lamb
2026-06-11 16:55:18 +01:00
parent 37b1ba8615 eefd5e1026
commit fd4629494b
236 changed files with 29059 additions and 1447 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/build.yml vendored
View File

@@ -26,7 +26,7 @@ env:
# staging repo (autumn-staging) -> us-east-1
# Branches allowed to deploy to staging via workflow_dispatch with tag=deploy-staging.
# Add short-lived PR branches here when you need staging without merging to dev.
STAGING_DEPLOY_BRANCH_ALLOWLIST: ""
STAGING_DEPLOY_BRANCH_ALLOWLIST: fix-health-check-redis-disabled-detection feat/track-rate-limit-redis feat/events-hourly-rollup fix/analytics-tz-bucket-offset uw-1-storage
jobs:
checks:

3
apps/leaf/package.json
View File

@@ -34,8 +34,7 @@
"e2b": "^2.8.4",
"hono": "4.12.7",
"postgres": "catalog:",
"zod": "^3.25.23",
"zod-v4": "npm:zod@^4.4.3"
"zod": "^3.25.23"
},
"devDependencies": {
"@ngrok/ngrok": "^1.7.0",

5
apps/leaf/src/lib/env.ts
View File

@@ -42,7 +42,10 @@ const envSchema = z
return {
...values,
MCP_SERVER_URL:
values.MCP_SERVER_URL ?? `http://localhost:${values.PORT}`,
values.MCP_SERVER_URL ??
(process.env.NODE_ENV === "production"
? "https://mcp.useautumn.com/mcp"
: `http://localhost:${values.PORT}`),
BETTER_AUTH_URL:
values.BETTER_AUTH_URL ??
(process.env.NODE_ENV === "production"

2
apps/leaf/src/main.ts
View File

@@ -18,6 +18,8 @@ app.use("*", async (c, next) => {
app.get("/health", (c) => c.json({ ok: true }));
app.route(
"",
createMcpRouter({

1
apps/leaf/tests/evals/harness/context/types.ts
View File

@@ -10,6 +10,7 @@ export type AutumnEvalToolName =
| "getCustomer"
| "getEntity"
| "getOrCreateCustomer"
| "getCurrentOrganization"
| "getPlan"
| "listCustomers"
| "listEntities"

18
apps/leaf/tests/evals/utils/scorers.ts
View File

@@ -196,9 +196,21 @@ const matchesApiCall = ({
actual.toolName === expected.toolName &&
(!expected.body || includesObject(actual.body, expected.body));
const valuesAtPath = ({ path, value }: { path: string; value: unknown }) => {
const valuesAtPath = ({
path,
value,
}: {
path: string;
value: unknown;
}): unknown[] => {
const parts = path.split(".");
const walk = ({ index, current }: { index: number; current: unknown }) => {
const walk = ({
index,
current,
}: {
index: number;
current: unknown;
}): unknown[] => {
if (index === parts.length) return [current];
const part = parts[index];
if (part === "*") {
@@ -343,7 +355,7 @@ export const expectedApiBodyNumberFields = ({
const values = valuesAtPath({ path, value: call.body });
return (
values.length > 0 &&
values.every((value) => typeof value === "number")
values.every((value: unknown) => typeof value === "number")
);
}),
)

4
apps/leaf/tests/harness/claudeCode.smoke.ts
View File

@@ -4,8 +4,8 @@ import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createSdkMcpServer, tool } from "@anthropic-ai/claude-agent-sdk";
// SDK tool() needs zod v4 internals; leaf is on zod 3, so use the aliased package.
import { z } from "zod-v4";
// SDK tool() needs v4 schemas from the same zod copy its peer resolves to (leaf's zod 3.25.x).
import { z } from "zod/v4";
import { createClaudeCodeHarness } from "../../src/harness/index.js";
import type {
HarnessEvent,

1684
bun.lock
View File

File diff suppressed because it is too large Load Diff

10
docker/Dockerfile
View File

@@ -41,13 +41,11 @@ COPY packages/stripe-sync/package.json packages/stripe-sync/
# install step doesn't fail before the real source is copied.
RUN mkdir -p scripts && touch scripts/preload-env.ts
# Install only the workspaces the runtime services need (server hosts workers +
# cron), plus their transitive workspace deps. --frozen-lockfile guarantees no
# re-resolution; --filter skips the frontend-heavy workspaces.
# Install the full workspace because runtime source imports cross package boundaries.
# --no-save keeps this image layer from mutating bun.lock.
RUN --mount=type=cache,target=/root/.bun/install/cache \
bun install --frozen-lockfile --ignore-scripts \
--filter @autumn/server \
--filter @autumn/leaf
bun install --ignore-scripts --no-save \
--minimum-release-age 0
FROM oven/bun:1.3.10
WORKDIR /app

3
packages/ai-sdk/package.json
View File

@@ -19,7 +19,7 @@
}
},
"scripts": {
"ts": "tsgo --noEmit --skipLibCheck",
"ts": "bunx tsgo --noEmit --skipLibCheck",
"test": "bun test tests/unit",
"build": "rm -rf dist && tsup",
"prepublishOnly": "bun run build"
@@ -33,6 +33,7 @@
},
"devDependencies": {
"@types/node": "^24.9.1",
"@typescript/native-preview": "catalog:",
"tsup": "^8.4.0",
"typescript": "^5.8.3"
}

4
packages/mcp/tests/utils/eval-test-utils.ts
View File

@@ -271,7 +271,7 @@ export const initMcpEval = ({
};
const generate = async (
message: string | string[],
maxSteps = leafChatAgentDefaults.maxSteps,
maxSteps: number = leafChatAgentDefaults.maxSteps,
) => {
messages.push({
role: "user",
@@ -293,7 +293,7 @@ export const initMcpEval = ({
generate,
approve: async (
message: string,
maxSteps = leafChatAgentDefaults.maxSteps,
maxSteps: number = leafChatAgentDefaults.maxSteps,
) => {
if (!pendingApproval) await generate(message, maxSteps);
if (!pendingApproval) {

38
scripts/dw/commands/setup.ts
View File

@@ -21,6 +21,42 @@ import { ensureEmulateRunning } from "../helpers/emulate.ts";
import { PROJECT_ROOT } from "../constants.ts";
import type { RegistryEntry } from "../types.ts";
function ensureAiSubmoduleSynced(): void {
const aiDir = `${PROJECT_ROOT}/ai`;
log("ensuring ai submodule is initialized");
const submoduleCode = shInherit(
"git",
["submodule", "update", "--init", "--recursive"],
{ cwd: PROJECT_ROOT },
);
if (submoduleCode !== 0) {
fatal(
`git submodule update --init --recursive failed (exit ${submoduleCode})`,
);
}
log("checking out ai submodule main branch");
const checkoutCode = shInherit("git", ["checkout", "main"], {
cwd: aiDir,
});
if (checkoutCode !== 0) {
fatal(`git checkout main failed in ai submodule (exit ${checkoutCode})`);
}
log("ensuring ai deps installed (bun install)");
const installCode = shInherit("bun", ["install"], { cwd: aiDir });
if (installCode !== 0) {
fatal(`bun install failed in ai submodule (exit ${installCode})`);
}
log("syncing ai skills");
const syncCode = shInherit("bun", ["sync"], { cwd: aiDir });
if (syncCode !== 0) {
fatal(`bun sync failed in ai submodule (exit ${syncCode})`);
}
}
export async function cmdSetup(): Promise<RegistryEntry> {
if (process.env.NODE_ENV === "production") {
fatal("bun dw is disabled in production");
@@ -32,6 +68,8 @@ export async function cmdSetup(): Promise<RegistryEntry> {
const installCode = shInherit("bun", ["install"], { cwd: PROJECT_ROOT });
if (installCode !== 0) fatal(`bun install failed (exit ${installCode})`);
ensureAiSubmoduleSynced();
const canonical = getCanonicalWorktree();
const cwd = getCurrentWorktree();
let registry = loadRegistry();

2
scripts/dw/constants.ts
View File

@@ -16,8 +16,6 @@ export const NEON_TEMPLATE_BRANCH = "dw-template";
export const NEON_PARENT_BRANCH = "production";
export const EMULATE_PID_FILE = join(homedir(), ".autumn-emulate.pid");
export const EMULATE_HEALTH_URL =
"https://google.emulate.localhost/.well-known/openid-configuration";
export const START_EMULATE_SH = join(SCRIPT_DIR, "../setup/start-emulate.sh");
export const ENV_LOCAL_TARGETS = [

12
scripts/dw/helpers/emulate.ts
View File

@@ -1,19 +1,17 @@
import { existsSync, readFileSync, rmSync } from "node:fs";
import { sh, log } from "./shell.ts";
import {
EMULATE_PID_FILE,
EMULATE_HEALTH_URL,
START_EMULATE_SH,
} from "../constants.ts";
import { EMULATE_PID_FILE, START_EMULATE_SH } from "../constants.ts";
import { portlessHttpsUrl } from "./ports.ts";
import { log, sh } from "./shell.ts";
function emulateReachable(): boolean {
const healthUrl = `${portlessHttpsUrl("google.emulate.localhost")}/.well-known/openid-configuration`;
const res = sh("curl", [
"-sf",
"-o",
"/dev/null",
"--max-time",
"1",
EMULATE_HEALTH_URL,
healthUrl,
]);
return res.code === 0;
}

56
scripts/dw/helpers/env-files.ts
View File

@@ -1,15 +1,34 @@
import { existsSync, readFileSync, renameSync, writeFileSync, rmSync } from "node:fs";
import { dirname, join } from "node:path";
import {
existsSync,
readFileSync,
renameSync,
rmSync,
writeFileSync,
} from "node:fs";
import { homedir } from "node:os";
import { dirname, join } from "node:path";
import {
ENV_LOCAL_DISABLED_SUFFIX,
ENV_LOCAL_TARGETS,
PROJECT_ROOT,
} from "../constants.ts";
import type { RegistryEntry } from "../types.ts";
import {
aliasesFor,
dragonflyPortFor,
elasticMqPortFor,
portlessHttpsUrl,
} from "./ports.ts";
import { log } from "./shell.ts";
import { forceSslVerifyFull } from "./url.ts";
import { aliasesFor, dragonflyPortFor, elasticMqPortFor } from "./ports.ts";
import { PROJECT_ROOT, ENV_LOCAL_TARGETS, ENV_LOCAL_DISABLED_SUFFIX } from "../constants.ts";
import type { RegistryEntry } from "../types.ts";
// Simple KEY=VALUE parse (no quoting/multiline). Sufficient for .env.local
// files we own end-to-end; preserves blank lines and comments untouched.
export function parseEnvFile(contents: string): { keys: string[]; values: Record<string, string>; raw: string[] } {
export function parseEnvFile(contents: string): {
keys: string[];
values: Record<string, string>;
raw: string[];
} {
const raw = contents.split(/\r?\n/);
const values: Record<string, string> = {};
const keys: string[] = [];
@@ -23,11 +42,14 @@ export function parseEnvFile(contents: string): { keys: string[]; values: Record
return { keys, values, raw };
}
export function mergeEnvFile(existing: string | null, managed: Record<string, string>): string {
export function mergeEnvFile(
existing: string | null,
managed: Record<string, string>,
): string {
if (!existing) {
return Object.entries(managed)
return `${Object.entries(managed)
.map(([k, v]) => `${k}=${v}`)
.join("\n") + "\n";
.join("\n")}\n`;
}
const parsed = parseEnvFile(existing);
const managedKeys = new Set(Object.keys(managed));
@@ -49,7 +71,7 @@ export function mergeEnvFile(existing: string | null, managed: Record<string, st
while (outLines.length > 0 && outLines[outLines.length - 1] === "") {
outLines.pop();
}
return outLines.join("\n") + "\n";
return `${outLines.join("\n")}\n`;
}
export function writeEnvLocalFiles(entry: RegistryEntry): void {
@@ -68,7 +90,7 @@ export function writeEnvLocalFiles(entry: RegistryEntry): void {
DATABASE_CRITICAL_URL: dbUrl,
BETTER_AUTH_URL: aliases.apiUrl,
CLIENT_URL: aliases.viteUrl,
EMULATE_GOOGLE_URL: "https://google.emulate.localhost",
EMULATE_GOOGLE_URL: portlessHttpsUrl("google.emulate.localhost"),
AUTUMN_TEST_BASE_URL: `http://localhost:${serverPort}`,
AUTUMN_TEST_VITE_URL: aliases.viteUrl,
STRIPE_WEBHOOK_SKIP_VERIFY: "true",
@@ -135,7 +157,11 @@ export function removeEnvLocalFiles(): void {
}
}
export function disableEnvLocalFiles(): { moved: number; missing: number; alreadyDisabled: number } {
export function disableEnvLocalFiles(): {
moved: number;
missing: number;
alreadyDisabled: number;
} {
let moved = 0;
let missing = 0;
let alreadyDisabled = 0;
@@ -156,7 +182,11 @@ export function disableEnvLocalFiles(): { moved: number; missing: number; alread
return { moved, missing, alreadyDisabled };
}
export function enableEnvLocalFiles(): { moved: number; missing: number; alreadyEnabled: number } {
export function enableEnvLocalFiles(): {
moved: number;
missing: number;
alreadyEnabled: number;
} {
let moved = 0;
let missing = 0;
let alreadyEnabled = 0;

34
scripts/dw/helpers/ports.ts
View File

@@ -1,5 +1,10 @@
import { sh, log } from "./shell.ts";
import { existsSync, readFileSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";
import type { WorktreeAliases } from "../types.ts";
import { log, sh } from "./shell.ts";
const PORTLESS_PROXY_PORT_FILE = join(homedir(), ".portless", "proxy.port");
export function dragonflyPortFor(worktreeNum: number): number {
return 6379 + (worktreeNum - 1) * 100;
@@ -28,17 +33,38 @@ export function aliasesFor(worktreeNum: number): WorktreeAliases {
const viteHost = `wt${worktreeNum}.localhost`;
return {
apiHost,
apiUrl: `https://${apiHost}`,
apiUrl: portlessHttpsUrl(apiHost),
viteHost,
viteUrl: `https://${viteHost}`,
viteUrl: portlessHttpsUrl(viteHost),
};
}
export function portlessHttpsUrl(host: string): string {
const port = currentPortlessProxyPort();
const suffix = port && port !== 443 ? `:${port}` : "";
return `https://${host}${suffix}`;
}
export function currentPortlessProxyPort(): number | undefined {
const envPort = Number(process.env.PORTLESS_PORT);
if (Number.isInteger(envPort) && envPort > 0) return envPort;
if (!existsSync(PORTLESS_PROXY_PORT_FILE)) return undefined;
const filePort = Number(
readFileSync(PORTLESS_PROXY_PORT_FILE, "utf-8").trim(),
);
if (Number.isInteger(filePort) && filePort > 0) return filePort;
return undefined;
}
export function killOwnPorts(worktreeNum: number): void {
const offset = (worktreeNum - 1) * 100;
const ports = [8080 + offset, 3000 + offset, 3001 + offset];
if (process.platform === "win32") return;
const lsof = sh("lsof", ports.flatMap((p) => ["-ti", `:${p}`]));
const lsof = sh(
"lsof",
ports.flatMap((p) => ["-ti", `:${p}`]),
);
const pids = lsof.stdout.split("\n").filter(Boolean);
for (const pid of pids) {
try {

22
scripts/dw/helpers/start.ts
View File

@@ -1,13 +1,13 @@
import { existsSync } from "node:fs";
import { join } from "node:path";
import { homedir } from "node:os";
import { log, fatal } from "./shell.ts";
import { registerPortlessAliases } from "./portless.ts";
import { rewriteDbEnv } from "./url.ts";
import { aliasesFor, killOwnPorts } from "./ports.ts";
import { tmuxSessionName, spawnDevInTmux } from "./tmux.ts";
import { join } from "node:path";
import { PROJECT_ROOT } from "../constants.ts";
import type { RegistryEntry } from "../types.ts";
import { registerPortlessAliases } from "./portless.ts";
import { portlessHttpsUrl } from "./ports.ts";
import { fatal, log } from "./shell.ts";
import { spawnDevInTmux, tmuxSessionName } from "./tmux.ts";
import { rewriteDbEnv } from "./url.ts";
export function buildDevEnvAndArgs(entry: RegistryEntry): {
env: Record<string, string>;
@@ -21,7 +21,7 @@ export function buildDevEnvAndArgs(entry: RegistryEntry): {
if (!databaseUrl) fatal("agent worktree missing databaseUrl");
env = rewriteDbEnv(env, databaseUrl);
if (!env.EMULATE_GOOGLE_URL) {
env.EMULATE_GOOGLE_URL = "https://google.emulate.localhost";
env.EMULATE_GOOGLE_URL = portlessHttpsUrl("google.emulate.localhost");
}
const portlessCa = join(homedir(), ".portless", "ca.pem");
if (existsSync(portlessCa) && !env.NODE_EXTRA_CA_CERTS) {
@@ -50,14 +50,18 @@ export function buildDevEnvAndArgs(entry: RegistryEntry): {
return { env, args };
}
export function startDev(entry: RegistryEntry, opts?: { allowTmux?: boolean }): never {
export function startDev(
entry: RegistryEntry,
opts?: { allowTmux?: boolean },
): never {
const { worktreeNum, branchName } = entry;
const { env, args } = buildDevEnvAndArgs(entry);
// Agent worktrees (N > 1) in a non-TTY invocation: wrap in detached tmux
// so the calling agent doesn't block. Canonical (N=1) stays inline always.
// Node/Bun sets isTTY to true when stdout is a TTY and undefined otherwise.
const useTmux = (opts?.allowTmux ?? true) && worktreeNum > 1 && !process.stdout.isTTY;
const useTmux =
(opts?.allowTmux ?? true) && worktreeNum > 1 && !process.stdout.isTTY;
if (useTmux) {
log(
`starting dev in tmux (worktree=${worktreeNum}${branchName ? `, branch=${branchName}` : ""}, non-TTY)`,

11
scripts/migrations/migrate-functions.ts
View File

@@ -3,6 +3,17 @@ import inquirer from "inquirer";
loadLocalEnv();
// Dev worktrees (scripts/dw): overlay server/.env.local -- the same override
// `bun dw run` gets via Bun's automatic .env.local loading -- so functions
// land on the worktree's Neon branch instead of the canonical dev DB. Never
// applied for prod targets (migrate-functions:prod): infisical injects the
// prod DATABASE_URL before this script starts, and prod URLs carry the
// us-east-2 marker (same convention as assertNotProductionDb).
if (!process.env.DATABASE_URL?.includes("us-east-2")) {
process.env.ENV_FILE = ".env.local";
loadLocalEnv({ force: true });
}
export const migrateFunctions = async () => {
// Dynamic import to ensure env is loaded first
const { initializeDatabaseFunctions } = await import(

6
scripts/package.json
View File

@@ -15,17 +15,17 @@
"chalk": "^5.3.0",
"dotenv": "^16.5.0",
"drizzle-orm": "catalog:",
"ink": "^5.1.0",
"ink": "^6.6.0",
"ioredis": "^5.10.0",
"inquirer": "^12.6.3",
"p-limit": "^7.2.0",
"pg": "8.20.0",
"react": "^18.3.1"
"react": "^19.2.1"
},
"devDependencies": {
"@types/bun": "^1.3.11",
"@types/pg": "8.20.0",
"@types/react": "^18.3.1",
"@types/react": "^19.2.1",
"tsx": "^4.19.2",
"typescript": "^5.7.3"
}

16
scripts/setup/start-emulate.sh
View File

@@ -8,13 +8,23 @@ ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
SEED="$ROOT/emulate.config.yaml"
LOG="$HOME/.autumn-emulate.log"
PID_FILE="$HOME/.autumn-emulate.pid"
PORTLESS_PORT_FILE="$HOME/.portless/proxy.port"
EMULATE_URL="https://google.emulate.localhost"
PORTLESS_PROXY_PORT="${PORTLESS_PORT:-}"
if [[ -z "$PORTLESS_PROXY_PORT" && -f "$PORTLESS_PORT_FILE" ]]; then
PORTLESS_PROXY_PORT="$(cat "$PORTLESS_PORT_FILE" 2>/dev/null || true)"
fi
if [[ -n "$PORTLESS_PROXY_PORT" && "$PORTLESS_PROXY_PORT" != "443" ]]; then
EMULATE_URL="${EMULATE_URL}:${PORTLESS_PROXY_PORT}"
fi
reachable() {
curl -sf -o /dev/null --max-time 1 "https://google.emulate.localhost/.well-known/openid-configuration"
curl -sf -o /dev/null --max-time 1 "${EMULATE_URL}/.well-known/openid-configuration"
}
if reachable; then
echo "[emulate] already reachable at https://google.emulate.localhost"
echo "[emulate] already reachable at ${EMULATE_URL}"
exit 0
fi
@@ -54,7 +64,7 @@ disown
# Block briefly until the emulator is actually serving so callers can race.
for _ in $(seq 1 30); do
if reachable; then
echo "[emulate] ready at https://google.emulate.localhost (pid $(cat "$PID_FILE"))"
echo "[emulate] ready at ${EMULATE_URL} (pid $(cat "$PID_FILE"))"
exit 0
fi
sleep 0.3

1
server/experiments/normalizedSubjectCacheExperiment.ts
View File

@@ -267,6 +267,7 @@ const generateNormalized = (): NormalizedFullSubject => {
},
},
rollovers: [] as any,
usage_windows: [] as any,
replaceables: [] as any,
customerPrice: null as any,
customerProductOptions: [] as any,

4
server/package.json
View File

@@ -159,8 +159,8 @@
"@types/mocha": "^10.0.10",
"@types/node": "^25.0.7",
"@types/pg": "8.20.0",
"@types/react": "18.3.28",
"@types/react-dom": "18.3.7",
"@types/react": "^19.2.17",
"@types/react-dom": "^19.2.3",
"@types/ws": "^8.18.1",
"artillery": "^2.0.30",
"cross-env": "^7.0.3",

80
server/src/_luaScriptsV2/fullSubject/rollUsageWindows/rollUsageWindows.lua Normal file
View File

@@ -0,0 +1,80 @@
--[[
Lua Script: Roll usage-window counters in a per-feature hash
Atomically patches rows in the reserved '_usage_windows' field: the lazy
roll (post-getFullSubject) zeroes counts whose window closed and advances
bounds/anchor to the current derivation. Atomicity matters because a
concurrent deduction may be writing the same field.
Fail-open: a missing/malformed field, or a row absent for a scope, is left
untouched (the write path creates rows; the roll only maintains them).
KEYS[1] = balance hash key
ARGV[1] = JSON params:
{
now: number,
ttl_seconds: number,
rolls: [{
internal_entity_id: string | null, -- scope selector
zero_usage: boolean, -- stored window closed: count dies
window_start_at: number,
window_end_at: number,
anchor_customer_entitlement_id: string | null,
}]
}
Returns JSON: { rolled: number }
]]
local params = cjson.decode(ARGV[1])
local now = safe_number(params.now)
local ttl_seconds = safe_number(params.ttl_seconds)
local USAGE_WINDOWS_FIELD = '_usage_windows'
local raw = redis.call('HGET', KEYS[1], USAGE_WINDOWS_FIELD)
if is_nil(raw) then
return cjson.encode({ rolled = 0 })
end
local ok, windows = pcall(cjson.decode, raw)
if not ok or type(windows) ~= 'table' then
return cjson.encode({ rolled = 0 })
end
local rolled = 0
for _, roll in ipairs(params.rolls or {}) do
local roll_entity = roll.internal_entity_id
for _, window in ipairs(windows) do
if type(window) == 'table' then
local window_entity = window.internal_entity_id
local entities_match =
(is_nil(roll_entity) and is_nil(window_entity))
or roll_entity == window_entity
if entities_match then
if roll.zero_usage then
window.usage = 0
end
window.window_start_at = roll.window_start_at
window.window_end_at = roll.window_end_at
window.anchor_customer_entitlement_id =
roll.anchor_customer_entitlement_id
window.updated_at = now
rolled = rolled + 1
end
end
end
end
if rolled == 0 then
return cjson.encode({ rolled = 0 })
end
local encoded = #windows > 0 and cjson.encode(windows) or '[]'
redis.call('HSET', KEYS[1], USAGE_WINDOWS_FIELD, encoded)
if ttl_seconds > 0 and redis.call('TTL', KEYS[1]) < 0 then
redis.call('EXPIRE', KEYS[1], ttl_seconds)
end
return cjson.encode({ rolled = rolled })

8
server/src/_luaScriptsV2/fullSubjectDeduction/contextUtilsV2.lua
View File

@@ -30,10 +30,18 @@ local function init_context(params)
local context = {
customer_entitlements = {},
rollovers = {},
-- Customer-scoped windowed-cap counters, loaded below alongside the other
-- subject state: { [feature_id] = { balance_key, windows, dirty } }.
usage_windows = read_usage_windows({
usage_window_limits = params.usage_window_limits,
balance_keys_by_feature_id = params.balance_keys_by_feature_id,
now = params.usage_window_now,
}),
org_id = params.org_id,
env = params.env,
customer_id = params.customer_id,
mutation_logs = {},
usage_window_mutations = {},
pending_writes = {},
pending_write_ids = {},
missing_customer_entitlement_ids =

86
server/src/_luaScriptsV2/fullSubjectDeduction/deductFromSubjectBalances.lua
View File

@@ -48,11 +48,30 @@
idempotency_ttl_ms: number | null
}
Usage windows (customer-scoped windowed caps):
CONFIG IN: params.usage_window_limits[] -- the resolved caps (limit,
bounds, dimension) from fullSubjectToUsageWindowLimits.
COUNTERS OUT: usage_windows_by_feature_id -- the post-deduction COUNTER
ROWS (DbUsageWindow: usage amounts, mirrors the
usage_windows table), NOT the config.
Counters live in the capped feature's balance hash under the reserved
'_usage_windows' field (so each capped feature's hash key must be in
KEYS[], via usageWindowFeatureIds in the TS key builder); they are loaded
into context.usage_windows by init_context and follow the same in-memory
mutate -> flush lifecycle as entitlement balances. Enforcement is woven
into the deduction passes like spend limits: each ent's deductible amount
is gated by window headroom (with credit conversions), and a window-capped
leftover flows through the standard overage_behaviour handling ('cap'
applies the partial deduction, 'reject' returns INSUFFICIENT_BALANCE). A
missing field loads as an empty counter set (fail open).
Returns JSON:
{
updates: { [cus_ent_id]: { balance, additional_balance, adjustment, entities, deducted, additional_deducted } },
rollover_updates: { [rollover_id]: { balance, usage, entities } },
modified_customer_entitlement_ids: string[],
usage_windows_by_feature_id: { [feature_id]: DbUsageWindow[] } | null,
usage_window_mutations: { usage_window_id, feature_id, internal_entity_id, window_start_at, usage_delta }[],
remaining: number,
error: string | null,
feature_id: string | null
@@ -111,6 +130,10 @@ local idempotency_ttl_ms = params.idempotency_ttl_ms
local lock = params.lock
local unwind_value = params.unwind_value
local lock_receipt_key = lock_receipt_key_from_keys
local usage_window_limits = params.usage_window_limits
local usage_window_now = params.usage_window_now
local usage_window_ttl_seconds = params.usage_window_ttl_seconds
local is_consumption = params.is_consumption
if not is_nil(idempotency_key) then
if redis.call('EXISTS', idempotency_key) == 1 then
@@ -139,11 +162,30 @@ if #customer_entitlement_deductions == 0 then
})
end
-- Usage windows are enforced for positive consumption INCLUDING locks (a
-- lock reserves headroom and counts at lock time), never for refunds,
-- target_balance, or granted-balance edits. Unwinds don't enforce but DO
-- load counters so the freed amount can be decremented back. Computed before
-- init_context so non-participating calls skip the counter reads entirely.
local has_usage_window_limits = not is_nil(usage_window_limits)
and #usage_window_limits > 0
-- A zero unwind_value (finalize at-or-above the lock) is no unwind at all:
-- the extra delta must still be enforced and counted.
local has_unwind = not is_nil(unwind_value) and safe_number(unwind_value) > 0
local enforce_usage_windows = is_consumption
and not has_unwind
and has_usage_window_limits
local unwind_usage_windows = has_unwind and has_usage_window_limits
local context = init_context({
org_id = org_id,
env = env,
customer_id = customer_id,
customer_entitlement_deductions = customer_entitlement_deductions,
usage_window_limits = (enforce_usage_windows or unwind_usage_windows)
and usage_window_limits
or nil,
usage_window_now = usage_window_now,
balance_keys_by_feature_id = params.balance_keys_by_feature_id,
debug = params.debug,
})
@@ -185,6 +227,14 @@ if not is_nil(unwind_value) and safe_number(unwind_value) > 0 then
-- Track which entitlements the unwind touched so the caller can sync them.
unwind_modified_cus_ent_ids = unwind_result.modified_customer_entitlement_ids or {}
if unwind_usage_windows then
decrement_usage_windows_for_unwind({
context = context,
iterations = unwind_result.iterations,
now = usage_window_now,
})
end
-- Fold any skipped unwind (missing entitlements/rollovers) into amount_to_deduct
-- so the forward pass compensates against current live entitlements.
local skipped = unwind_result.remaining_signed_unwind_value or 0
@@ -194,6 +244,7 @@ if not is_nil(unwind_value) and safe_number(unwind_value) > 0 then
end
local logger = context.logger
logger.log("=== LUA DEDUCTION START ===")
logger.log("=== PARAMS ===")
logger.log(" amount_to_deduct: %s", tostring(amount_to_deduct or "nil"))
@@ -235,18 +286,21 @@ for _, cus_ent_id in ipairs(unwind_modified_cus_ent_ids) do
end
end
local modified_customer_entitlement_ids = collect_modified_customer_entitlement_ids({
context = context,
extra_customer_entitlement_ids = unwind_modified_cus_ent_ids,
})
logger.log(" remaining_amount: %s", tostring(remaining_amount or "nil"))
logger.log(" is_refund: %s", tostring(remaining_amount < 0 or false))
local mutation_logs = context.mutation_logs
if type(mutation_logs) ~= 'table' or #mutation_logs == 0 then
mutation_logs = cjson.decode('[]')
end
-- Throw error and don't apply updates if we're in reject mode and there's still remaining amount
local usage_window_mutations = context.usage_window_mutations
if type(usage_window_mutations) ~= 'table' or #usage_window_mutations == 0 then
usage_window_mutations = cjson.decode('[]')
end
-- Throw error and don't apply updates if we're in reject mode and there's
-- still remaining amount. Usage-window shortfalls flow through here like any
-- other: the deduction passes already gated every ent by window headroom, so
-- a window-capped leftover clamps under 'cap' and rejects as
-- INSUFFICIENT_BALANCE under 'reject'.
if remaining_amount > 0 and overage_behaviour == 'reject' then
return cjson.encode({
error = 'INSUFFICIENT_BALANCE',
@@ -259,6 +313,19 @@ if remaining_amount > 0 and overage_behaviour == 'reject' then
})
end
if enforce_usage_windows then
increment_usage_window_counters({
context = context,
usage_window_limits = usage_window_limits,
now = usage_window_now,
})
end
local modified_customer_entitlement_ids = collect_modified_customer_entitlement_ids({
context = context,
extra_customer_entitlement_ids = unwind_modified_cus_ent_ids,
})
if not is_nil(lock)
and not is_nil(lock.enabled)
and lock.enabled
@@ -308,6 +375,10 @@ update_aggregated_balances({
mutation_logs = mutation_logs,
})
if enforce_usage_windows or unwind_usage_windows then
apply_usage_window_writes(context, usage_window_ttl_seconds)
end
if not is_nil(idempotency_key) and not is_nil(idempotency_ttl_ms) then
redis.call('SET', idempotency_key, '1', 'PX', idempotency_ttl_ms)
end
@@ -319,6 +390,9 @@ return cjson.encode({
rollover_updates = rollover_updates,
modified_customer_entitlement_ids = modified_customer_entitlement_ids,
mutation_logs = mutation_logs,
usage_windows_by_feature_id =
usage_windows_to_result(context) or cjson.null,
usage_window_mutations = usage_window_mutations,
remaining = remaining_amount,
error = cjson.null,
logs = context.logs

3
server/src/_luaScriptsV2/fullSubjectDeduction/lock/unwindLockV2.lua
View File

@@ -442,5 +442,8 @@ local function unwind_lock_on_context(params)
modified_customer_entitlement_ids = modified_ids.modified_customer_entitlement_ids,
modified_rollover_ids = modified_ids.modified_rollover_ids,
mutation_logs = context.mutation_logs,
-- Per-item applied amounts (tracked units + credit_cost), so callers can
-- mirror the unwind onto usage-window counters.
iterations = unwind_items_result.iterations,
}
end

29
server/src/_luaScriptsV2/fullSubjectDeduction/readSubjectBalances.lua
View File

@@ -18,28 +18,45 @@ local function read_subject_balances(params)
local balances_by_id = {}
local missing_customer_entitlement_ids = {}
local entries_by_balance_key = {}
local seen_ids_by_balance_key = {}
local balance_keys_by_feature_id = safe_table(params.balance_keys_by_feature_id)
for _, ent_obj in ipairs(params.customer_entitlement_deductions or {}) do
local customer_entitlement_id = ent_obj.customer_entitlement_id
local feature_id = ent_obj.feature_id
local function queue_balance_read(customer_entitlement_id, feature_id)
if not (customer_entitlement_id and feature_id) then
return false
end
if customer_entitlement_id and feature_id then
local balance_key = balance_keys_by_feature_id[feature_id]
if not balance_key then
table.insert(missing_customer_entitlement_ids, customer_entitlement_id)
else
return false
end
if entries_by_balance_key[balance_key] == nil then
entries_by_balance_key[balance_key] = {
feature_id = feature_id,
customer_entitlement_ids = {},
}
seen_ids_by_balance_key[balance_key] = {}
end
if seen_ids_by_balance_key[balance_key][customer_entitlement_id] then
return true
end
seen_ids_by_balance_key[balance_key][customer_entitlement_id] = true
table.insert(
entries_by_balance_key[balance_key].customer_entitlement_ids,
customer_entitlement_id
)
return true
end
for _, ent_obj in ipairs(params.customer_entitlement_deductions or {}) do
local customer_entitlement_id = ent_obj.customer_entitlement_id
if customer_entitlement_id then
local queued = queue_balance_read(customer_entitlement_id, ent_obj.feature_id)
if not queued then
table.insert(missing_customer_entitlement_ids, customer_entitlement_id)
end
end
end

77
server/src/_luaScriptsV2/fullSubjectDeduction/runDeductionOnContextV2.lua
View File

@@ -50,7 +50,7 @@ local function process_deduction_pass(params)
local ent_id = ent_obj.customer_entitlement_id
local credit_cost = ent_obj.credit_cost
local ent_feature_id = ent_obj.feature_id
if credit_cost == cjson.null or credit_cost == nil then
if credit_cost == cjson.null or credit_cost == nil or credit_cost == 0 then
credit_cost = 1
end
@@ -85,23 +85,43 @@ local function process_deduction_pass(params)
usage_allowed = usage_allowed or overage_behavior_is_allow
local should_process = not skip_if_not_usage_allowed or usage_allowed
local skip_reason = "usage_allowed=false"
if not context.customer_entitlements[ent_id] then
should_process = false
skip_reason = "not in context"
end
-- Usage-window gate, mirroring the spend-limit overage gate above: cap
-- this ent's deductible amount by the remaining window headroom (metered
-- limits cap every ent in tracked units; balance limits cap ents of the
-- capped feature, converted via THIS ent's credit_cost). A fully blocked
-- ent is skipped rather than breaking the loop -- a balance-dim cap only
-- binds its own feature's pools, so other ents may be unconstrained.
local ent_amount = remaining_amount
if should_process and remaining_amount > 0 then
local available_from_usage_windows = get_available_from_usage_windows({
context = context,
ent_feature_id = ent_feature_id,
credit_cost = credit_cost,
})
if not is_nil(available_from_usage_windows)
and available_from_usage_windows < ent_amount then
ent_amount = available_from_usage_windows
end
if ent_amount == 0 then
should_process = false
skip_reason = "usage window headroom exhausted"
end
end
if not should_process then
logger.log("%s skipping %s - usage_allowed=false or not in context", pass_name, ent_id)
elseif credit_cost == 0 then
-- Zero credit cost (e.g. -100% markup AI model): the usage is free.
-- Consume the requested amount without touching any balance.
logger.log("%s ent %s credit_cost=0 - free deduction, no balance change", pass_name, ent_id)
remaining_amount = 0
logger.log("%s skipping %s - %s", pass_name, ent_id, skip_reason)
else
local deducted = deduct_from_main_balance({
context = context,
ent_id = ent_id,
target_entity_id = target_entity_id,
amount = remaining_amount,
amount = ent_amount,
credit_cost = credit_cost,
pass_number = pass_number,
available_overage = available_overage,
@@ -112,7 +132,17 @@ local function process_deduction_pass(params)
log_prefix = pass_name,
})
remaining_amount = remaining_amount - (deducted / credit_cost)
local deducted_units = deducted / credit_cost
remaining_amount = remaining_amount - deducted_units
-- Settle the gate: record what this ent actually drained against every
-- applicable window limit so the next ent sees the reduced headroom.
consume_usage_window_headroom({
context = context,
ent_feature_id = ent_feature_id,
credit_cost = credit_cost,
units = deducted_units,
})
if deducted ~= 0 then
if not updates[ent_id] then
@@ -151,6 +181,26 @@ local function process_rollover_deduction(params)
return 0
end
-- Metered window limits count tracked units regardless of funding source,
-- so they gate the rollover phase too. Balance limits do not (ent_feature_id
-- = nil): rollover drains stay outside credit-pool caps, matching how spend
-- limits ignore them.
local rollover_amount = remaining_amount
local available_from_usage_windows = get_available_from_usage_windows({
context = context,
ent_feature_id = nil,
credit_cost = 1,
})
if not is_nil(available_from_usage_windows)
and available_from_usage_windows < rollover_amount then
rollover_amount = available_from_usage_windows
end
if rollover_amount <= 0 then
logger.log("Rollover deduction skipped - usage window headroom exhausted")
return 0
end
local first_ent = customer_entitlement_deductions[1]
local has_entity_scope = false
if first_ent then
@@ -160,11 +210,18 @@ local function process_rollover_deduction(params)
local rollover_deducted = deduct_from_rollovers({
context = context,
rollovers = rollovers,
amount = remaining_amount,
amount = rollover_amount,
target_entity_id = target_entity_id,
has_entity_scope = has_entity_scope,
})
consume_usage_window_headroom({
context = context,
ent_feature_id = nil,
credit_cost = 1,
units = rollover_deducted,
})
logger.log("Rollover deduction: deducted=%s, remaining=%s", rollover_deducted, remaining_amount - rollover_deducted)
return rollover_deducted

101
server/src/_luaScriptsV2/fullSubjectDeduction/usageWindows/readUsageWindows.lua Normal file
View File

@@ -0,0 +1,101 @@
-- ============================================================================
-- READ USAGE WINDOWS
-- Loads customer-scoped usage-window counter state into the deduction context
-- (sibling of read_subject_balances). Counters live in the capped feature's
-- balance hash under the reserved '_usage_windows' field, as a lean ARRAY of
-- rows mirroring the usage_windows table (DbUsageWindow):
-- { id, internal_customer_id, internal_entity_id, feature_id,
-- internal_feature_id, anchor_customer_entitlement_id,
-- window_start_at, window_end_at, usage, updated_at }
--
-- Each loaded entry also carries the deduction-time runtime state the per-ent
-- gate consumes: the resolved limit, its dimension, the remaining `headroom`
-- (limit - current window usage, decremented as the deduction passes drain
-- it), and `consumed` (this operation's total, in the limit's native unit).
--
-- FAIL OPEN: a missing/undecodable field (or an undeclared balance key) loads
-- as an empty counter set -- the window simply restarts. Stale-cache guards
-- may return in a future iteration.
-- ============================================================================
local USAGE_WINDOWS_FIELD = '_usage_windows'
-- ONE mutable counter row per scope: a row matches its limit on
-- internal_entity_id alone. Bounds are payload, not identity.
local function find_usage_window(windows, limit)
local limit_entity = limit.internal_entity_id
for _, window in ipairs(windows) do
if type(window) == 'table' then
local window_entity = window.internal_entity_id
local entities_match =
(is_nil(limit_entity) and is_nil(window_entity))
or limit_entity == window_entity
if entities_match then
return window
end
end
end
return nil
end
-- Returns { [feature_id] = { balance_key, windows, dirty, limit,
-- dimension_type, headroom, consumed } }, one entry per distinct capped
-- feature in usage_window_limits.
local function read_usage_windows(params)
local limits = params.usage_window_limits or {}
local balance_keys_by_feature_id =
safe_table(params.balance_keys_by_feature_id)
local usage_windows = {}
for _, limit in ipairs(limits) do
local feature_id = limit.feature_id
if usage_windows[feature_id] == nil then
local balance_key = balance_keys_by_feature_id[feature_id]
local windows = nil
if not is_nil(balance_key) then
local raw_value = redis.call('HGET', balance_key, USAGE_WINDOWS_FIELD)
windows = safe_decode(raw_value)
end
if type(windows) ~= 'table' then
windows = new_empty_array()
end
-- cjson decodes an empty JSON object ({}) to the same empty table as [];
-- a non-empty map-like blob should be impossible for this field, but
-- reset it defensively rather than letting ipairs skip rows silently.
if next(windows) ~= nil and windows[1] == nil then
windows = new_empty_array()
end
local existing = find_usage_window(windows, limit)
-- A count is valid only within its exact stamped window: derive 0 when
-- it expired OR its bounds no longer match the current derivation (the
-- lazy roll persists the zero; this read must not trust it blindly).
local current_usage = 0
if not is_nil(existing)
and safe_number(existing.window_end_at) > safe_number(params.now)
and safe_number(existing.window_start_at) == limit.window_start_at
then
current_usage = safe_number(existing.usage)
end
local headroom = safe_number(limit.limit) - current_usage
if headroom < 0 then
headroom = 0
end
usage_windows[feature_id] = {
balance_key = not is_nil(balance_key) and balance_key or nil,
windows = windows,
dirty = false,
limit = limit,
dimension_type = limit.dimension_type,
headroom = headroom,
consumed = 0,
}
end
end
return usage_windows
end

301
server/src/_luaScriptsV2/fullSubjectDeduction/usageWindows/usageWindowContextUtilsV2.lua Normal file
View File

@@ -0,0 +1,301 @@
-- ============================================================================
-- USAGE WINDOW CONTEXT UTILITIES (V2)
-- Hard windowed usage-limit enforcement against context.usage_windows (loaded
-- by init_context via read_usage_windows), integrated into the deduction
-- passes the same way spend limits are:
-- per-ent gate (get_available_from_usage_windows, in the deduction loop)
-- -> consume headroom as each ent drains (consume_usage_window_headroom)
-- -> update_in_memory_usage_window (mark dirty) -> apply_usage_window_writes.
-- A window-capped leftover is handled by the standard overage_behaviour path
-- ('cap' applies the partial deduction, 'reject' returns INSUFFICIENT_BALANCE)
-- -- no window-specific error.
--
-- CONFIG IN: usage_window_limits[] -- resolved caps (limit, bounds,
-- dimension) from fullSubjectToUsageWindowLimits.
-- COUNTERS OUT: context.usage_windows[feature_id].windows -- DbUsageWindow
-- rows (usage amounts), NOT the config.
--
-- Units: the deduction loop works in TRACKED-FEATURE UNITS; each ent's
-- credit_cost converts them to that ent's balance units. A metered_feature
-- limit counts tracked units (applies to every ent in the deduction set); a
-- balance limit counts credits drained from ents OF the capped feature, so
-- headroom converts via credit_cost at the gate.
-- ============================================================================
-- Tolerance for float drift (credit-ratio conversions leave sub-nano noise).
local USAGE_WINDOW_EPSILON = 1e-9
-- Max tracked units deductible from ONE ent given every applicable window
-- limit, or nil when unbounded. Windows never store a conversion -- headroom
-- lives in the limit's own unit (tracked units for metered dims, credits for
-- balance dims) and is converted HERE, per call, with the calling ent's
-- credit_cost: the same balance-dim headroom yields different unit allowances
-- for ents with different credit ratios, and only ents OF the capped feature
-- are bound by it at all. Metered dims need no conversion (the deduction loop
-- is denominated in tracked units, whatever pool funds them).
--
-- Pass ent_feature_id = nil for the rollover phase: metered limits still
-- apply (rollover drains consume tracked units), balance limits do not
-- (parity with spend limits, whose overage math also ignores rollover
-- drains).
local function get_available_from_usage_windows(params)
local context = params.context
local ent_feature_id = params.ent_feature_id
local credit_cost = params.credit_cost or 1
local allowed = nil
for feature_id, feature_windows in pairs(context.usage_windows or {}) do
local headroom = feature_windows.headroom
if headroom <= USAGE_WINDOW_EPSILON then
headroom = 0
end
local units = nil
if feature_windows.dimension_type ~= 'balance' then
units = headroom
elseif not is_nil(ent_feature_id) and feature_id == ent_feature_id then
units = headroom / credit_cost
end
if units ~= nil and (allowed == nil or units < allowed) then
allowed = units
end
end
return allowed
end
-- Records `units` tracked units drained from an ent against every applicable
-- limit: metered limits consume units 1:1, balance limits consume
-- units * credit_cost (credits). Decrements live headroom so the next ent's
-- gate sees it, and accumulates `consumed` for the counter increment.
local function consume_usage_window_headroom(params)
local context = params.context
local ent_feature_id = params.ent_feature_id
local credit_cost = params.credit_cost or 1
local units = params.units or 0
if units <= 0 then
return
end
for feature_id, feature_windows in pairs(context.usage_windows or {}) do
local consumed = nil
if feature_windows.dimension_type ~= 'balance' then
consumed = units
elseif not is_nil(ent_feature_id) and feature_id == ent_feature_id then
consumed = units * credit_cost
end
if consumed ~= nil and consumed > 0 then
feature_windows.headroom = feature_windows.headroom - consumed
if feature_windows.headroom < 0 then
feature_windows.headroom = 0
end
feature_windows.consumed = feature_windows.consumed + consumed
end
end
end
-- Sibling of append_mutation_log: records which window row moved and by how
-- much (usage_delta in the limit's native unit). Kept as its own stream so
-- mutation_logs stays entitlement/rollover-shaped.
local function append_usage_window_mutation(params)
local context = params.context
table.insert(context.usage_window_mutations, {
usage_window_id = params.usage_window_id or cjson.null,
feature_id = params.feature_id,
internal_entity_id = params.internal_entity_id or cjson.null,
window_start_at = params.window_start_at,
usage_delta = params.usage_delta or 0,
})
end
-- In-memory mutation for one limit (sibling of
-- update_in_memory_customer_entitlement_mutation). ONE mutable row per scope:
-- zero the count if its stored window closed (defensive guard -- the lazy
-- roll action owns the roll), stamp the current bounds/anchor, add consumed.
local function update_in_memory_usage_window(params)
local context = params.context
local limit = params.limit
local now = params.now
local feature_windows = context.usage_windows[limit.feature_id]
if feature_windows == nil then
return
end
if feature_windows.consumed > USAGE_WINDOW_EPSILON then
local existing = find_usage_window(feature_windows.windows, limit)
if is_nil(existing) then
-- The TS-minted candidate id is used ONLY at creation; under concurrency
-- the second request finds the first one's row and its id is discarded.
existing = {
id = limit.new_window_id,
internal_customer_id = limit.internal_customer_id,
internal_entity_id = limit.internal_entity_id,
feature_id = limit.feature_id,
internal_feature_id = limit.internal_feature_id,
usage = 0,
}
table.insert(feature_windows.windows, existing)
elseif safe_number(existing.window_end_at) <= now
or safe_number(existing.window_start_at) ~= limit.window_start_at
then
-- A count never survives its stamped window: zero on expiry AND on any
-- bounds re-derivation mismatch (plan change).
existing.usage = 0
end
existing.window_start_at = limit.window_start_at
existing.window_end_at = limit.window_end_at
existing.anchor_customer_entitlement_id =
limit.anchor_customer_entitlement_id
existing.usage = safe_number(existing.usage) + feature_windows.consumed
existing.updated_at = now
feature_windows.dirty = true
append_usage_window_mutation({
context = context,
usage_window_id = existing.id,
feature_id = limit.feature_id,
internal_entity_id = limit.internal_entity_id,
window_start_at = limit.window_start_at,
usage_delta = feature_windows.consumed,
})
end
end
-- Applies each limit's in-flight consumed amount to its counter row.
-- Mirrors a lock UNWIND onto the counters: each applied unwind iteration
-- frees window headroom (metered dims by tracked units, balance dims by the
-- credits restored to that feature's entitlements). Clamped at 0; only the
-- limit's CURRENT window is decremented (a roll between lock and unwind
-- forfeits the old window's count, which is the conservative outcome).
local function decrement_usage_windows_for_unwind(params)
local context = params.context
local iterations = safe_table(params.iterations)
local now = params.now
if is_nil(context.usage_windows) or #iterations == 0 then
return
end
local total_units = 0
local credits_by_feature_id = {}
for _, iteration in ipairs(iterations) do
local units = safe_number(iteration.unwind_iteration_value)
total_units = total_units + units
local item = iteration.item or {}
local ent_feature_id = nil
local ent = context.customer_entitlements[item.customer_entitlement_id]
if ent then
ent_feature_id = ent.feature_id
elseif item.rollover_id and context.rollovers[item.rollover_id] then
local rollover_ent = context.customer_entitlements[
context.rollovers[item.rollover_id].cus_ent_id
]
if rollover_ent then
ent_feature_id = rollover_ent.feature_id
end
end
if ent_feature_id then
local credits = units * safe_number(item.credit_cost or 1)
credits_by_feature_id[ent_feature_id] =
(credits_by_feature_id[ent_feature_id] or 0) + credits
end
end
for feature_id, feature_windows in pairs(context.usage_windows) do
local amount = 0
if feature_windows.dimension_type == 'balance' then
amount = credits_by_feature_id[feature_id] or 0
else
amount = total_units
end
if amount > 0 then
local existing = find_usage_window(
feature_windows.windows,
feature_windows.limit
)
if not is_nil(existing) then
local current = safe_number(existing.usage)
local next_usage = current - amount
if next_usage < 0 then
next_usage = 0
end
if next_usage ~= current then
existing.usage = next_usage
existing.updated_at = now
feature_windows.dirty = true
append_usage_window_mutation({
context = context,
usage_window_id = existing.id,
feature_id = feature_windows.limit.feature_id,
internal_entity_id = feature_windows.limit.internal_entity_id,
window_start_at = feature_windows.limit.window_start_at,
usage_delta = next_usage - current,
})
end
end
end
end
end
local function increment_usage_window_counters(params)
local context = params.context
local limits = params.usage_window_limits or {}
for _, limit in ipairs(limits) do
update_in_memory_usage_window({
context = context,
limit = limit,
now = params.now,
})
end
end
-- Persists dirty counter arrays back to their '_usage_windows' fields
-- (sibling of apply_pending_writes; direct HSET like updateAggregatedBalances
-- since the cusEnt pending-write path is keyed by entitlement blobs).
-- The EXPIRE guard is load-bearing under fail-open: a write to a hash that
-- did not exist (capped feature with no entitlements and no rebuild yet)
-- must not create an immortal key.
local function apply_usage_window_writes(context, ttl_seconds)
local ttl = tonumber(ttl_seconds)
for _, feature_windows in pairs(context.usage_windows or {}) do
if feature_windows.dirty and not is_nil(feature_windows.balance_key) then
redis.call(
'HSET',
feature_windows.balance_key,
USAGE_WINDOWS_FIELD,
cjson.encode(feature_windows.windows)
)
if ttl and ttl > 0
and redis.call('TTL', feature_windows.balance_key) < 0 then
redis.call('EXPIRE', feature_windows.balance_key, ttl)
end
end
end
end
-- Result payload: { [feature_id] = windows[] } for every loaded capped
-- feature, so the TS caller can refresh the in-flight subject and hand the
-- post-deduction counters to syncItemV4 (no Redis re-read).
local function usage_windows_to_result(context)
local result = nil
for feature_id, feature_windows in pairs(context.usage_windows or {}) do
if result == nil then
result = {}
end
result[feature_id] = feature_windows.windows
end
return result
end

13
server/src/_luaScriptsV2/luaScriptsV2.ts
View File

@@ -44,11 +44,14 @@ import READ_SUBJECT_BALANCES from "./fullSubjectDeduction/readSubjectBalances.lu
import RUN_DEDUCTION_ON_CONTEXT_V2 from "./fullSubjectDeduction/runDeductionOnContextV2.lua";
import SPEND_LIMIT_UTILS_V2 from "./fullSubjectDeduction/spendLimitUtilsV2.lua";
import UPDATE_AGGREGATED_BALANCES from "./fullSubjectDeduction/updateAggregatedBalances.lua";
import READ_USAGE_WINDOWS from "./fullSubjectDeduction/usageWindows/readUsageWindows.lua";
import USAGE_WINDOW_CONTEXT_UTILS_V2 from "./fullSubjectDeduction/usageWindows/usageWindowContextUtilsV2.lua";
// ============================================================================
// UPDATE SUBJECT BALANCES HELPERS (V2 cache — per-feature hash updates)
// ============================================================================
import ROLL_USAGE_WINDOWS_MAIN from "./fullSubject/rollUsageWindows/rollUsageWindows.lua";
import APPLY_FIELD_UPDATES from "./fullSubject/updateSubjectBalances/applyFieldUpdates.lua";
import UPDATE_CONTEXT_UTILS from "./fullSubject/updateSubjectBalances/updateContextUtils.lua";
import UPDATE_SUBJECT_BALANCES_MAIN from "./fullSubject/updateSubjectBalances/updateSubjectBalances.lua";
@@ -200,11 +203,13 @@ export const UPDATE_CUSTOMER_PRODUCT_SCRIPT =
*/
export const DEDUCT_FROM_SUBJECT_BALANCES_SCRIPT = `${LUA_UTILS}
${READ_SUBJECT_BALANCES}
${READ_USAGE_WINDOWS}
${CONTEXT_UTILS_V2}
${GET_TOTAL_BALANCE}
${DEDUCT_FROM_ROLLOVERS_V2}
${DEDUCT_FROM_MAIN_BALANCE_V2}
${SPEND_LIMIT_UTILS_V2}
${USAGE_WINDOW_CONTEXT_UTILS_V2}
${RUN_DEDUCTION_ON_CONTEXT_V2}
${MUTATION_ITEM_UTILS}
${LOCK_RECEIPT_UTILS_V2}
@@ -238,3 +243,11 @@ ${UPDATE_CONTEXT_UTILS}
${APPLY_FIELD_UPDATES}
${UPDATE_AGGREGATED_BALANCES}
${UPDATE_SUBJECT_BALANCES_MAIN}`;
/**
* Lua script for atomically rolling usage-window counters in a per-feature
* balance hash's '_usage_windows' field (zero expired counts, advance
* bounds/anchor). Called once per feature via pipeline by the lazy roll.
*/
export const ROLL_USAGE_WINDOWS_SCRIPT = `${LUA_UTILS}
${ROLL_USAGE_WINDOWS_MAIN}`;

2
server/src/db/shed503OnTransientError.ts
View File

@@ -18,7 +18,7 @@ export const shed503OnTransientError = async <T>({
if (!(isTransientDbError({ error }) || isTransientRedisError({ error }))) {
throw error;
}
ctx.logger.warn(`[${source}] DB unavailable, shedding with 503`, {
ctx.logger.warn(`[${source}] transient DB error, shedding with 503`, {
type: `${source}_fail_open`,
error,
});

16
server/src/external/redis/initUtils/createRedisAvailability.ts vendored
View File

@@ -114,16 +114,12 @@ export const createRedisAvailability = ({
}
const shouldReconnectReadyClient =
failedWhileReady &&
consecutiveFailures + 1 >= REDIS_FAILURES_TO_DEGRADE;
failedWhileReady && consecutiveFailures + 1 >= REDIS_FAILURES_TO_DEGRADE;
if (shouldReconnectReadyClient) {
await reconnectRedis();
} else if (redis.status !== "ready") {
if (
redis.status === "connecting" ||
redis.status === "reconnecting"
) {
if (redis.status === "connecting" || redis.status === "reconnecting") {
reconnectStartedAt ??= Date.now();
if (Date.now() - reconnectStartedAt < REDIS_STALE_RECONNECT_MS) {
return false;
@@ -149,10 +145,7 @@ export const createRedisAvailability = ({
return {
prime: async () => {
if (!hasConfig) return;
if (
redis.status === "connecting" ||
redis.status === "reconnecting"
) {
if (redis.status === "connecting" || redis.status === "reconnecting") {
await waitForRedisReady(redis, logPrefix).catch(() => undefined);
}
const available = await probeRedisAvailability();
@@ -174,8 +167,7 @@ export const createRedisAvailability = ({
clearInterval(redisMonitorInterval);
redisMonitorInterval = null;
},
shouldUseRedis: () =>
hasConfig && redisAvailabilityState === "healthy",
shouldUseRedis: () => hasConfig && redisAvailabilityState === "healthy",
getRedisAvailability: (): RedisAvailabilitySnapshot => ({
configured: hasConfig,
state: redisAvailabilityState,

2
server/src/external/redis/initUtils/redisAvailability.ts vendored
View File

@@ -1,8 +1,8 @@
import { redis } from "./redisClientRegistry.js";
import {
createRedisAvailability,
type RedisAvailabilitySnapshot,
} from "./createRedisAvailability.js";
import { redis } from "./redisClientRegistry.js";
import { hasRedisConfig } from "./redisConfig.js";
const redisAvailability = createRedisAvailability({

1
server/src/external/redis/initUtils/redisTypes.ts vendored
View File

@@ -92,6 +92,7 @@ declare module "ioredis" {
balanceKey: string,
paramsJson: string,
): Promise<string>;
rollUsageWindows(balanceKey: string, paramsJson: string): Promise<string>;
deleteFullCustomerCache(
cacheKey: string,
testGuardKey: string,

6
server/src/external/redis/initUtils/registerRedisCommands.ts vendored
View File

@@ -22,6 +22,7 @@ import {
DEDUCT_FROM_SUBJECT_BALANCES_SCRIPT,
DELETE_FULL_CUSTOMER_CACHE_SCRIPT,
RESET_CUSTOMER_ENTITLEMENTS_SCRIPT,
ROLL_USAGE_WINDOWS_SCRIPT,
SET_CACHED_FULL_SUBJECT_SCRIPT,
SET_FULL_CUSTOMER_CACHE_SCRIPT,
UPDATE_CACHED_INVOICE_V2_SCRIPT,
@@ -127,6 +128,11 @@ export const registerRedisCommands = ({
lua: prepareScript(UPDATE_SUBJECT_BALANCES_SCRIPT),
});
redisInstance.defineCommand("rollUsageWindows", {
numberOfKeys: 1,
lua: prepareScript(ROLL_USAGE_WINDOWS_SCRIPT),
});
redisInstance.defineCommand("deleteFullCustomerCache", {
numberOfKeys: 4,
lua: DELETE_FULL_CUSTOMER_CACHE_SCRIPT,

12
server/src/external/redis/utils/withRedisFailOpen.ts vendored
View File

@@ -3,16 +3,18 @@ import { shouldUseRedisV2 } from "@/external/redis/initUtils/redisV2Availability
import { RedisUnavailableError } from "./errors.js";
import { isTransientRedisError } from "./isTransientRedisError.js";
/** Runs `run`. If Redis is unavailable or a transient DB error occurs,
* calls `fallback`. Any other error propagates. */
/** Runs `run`. If Redis is unavailable, a transient DB error occurs, or
* `alsoFailOpen` matches, calls `fallback`. Any other error propagates. */
export const withRedisFailOpen = async <T>({
source,
run,
fallback,
alsoFailOpen,
}: {
source: string;
run: () => T | Promise<T>;
fallback: (error: unknown) => T | Promise<T>;
alsoFailOpen?: (error: unknown) => boolean;
}): Promise<T> => {
try {
if (!shouldUseRedisV2()) {
@@ -21,7 +23,11 @@ export const withRedisFailOpen = async <T>({
return await run();
} catch (error) {
if (isTransientRedisError({ error }) || isTransientDbError({ error })) {
if (
isTransientRedisError({ error }) ||
isTransientDbError({ error }) ||
alsoFailOpen?.(error)
) {
return await fallback(error);
}

6
server/src/external/stripe/priceToStripeItem/priceToUsageInAdvance.ts vendored
View File

@@ -1,4 +1,5 @@
import {
atmnToStripeAmount,
type EntitlementWithFeature,
type FeatureOptions,
featureOptionUtils,
@@ -58,7 +59,10 @@ export const priceToOneOffAndTiered = ({
product: config.stripe_product_id
? config.stripe_product_id
: stripeProductId,
unit_amount: Number(amount.toFixed(2)) * 100,
unit_amount: atmnToStripeAmount({
amount,
currency: orgToCurrency({ org }),
}),
currency: orgToCurrency({ org }),
},

3
server/src/external/stripe/stripePriceUtils.ts vendored
View File

@@ -1,4 +1,5 @@
import {
atmnToStripeAmount,
BillingInterval,
type Customer,
type Feature,
@@ -95,7 +96,7 @@ export const getInvoiceItemForUsage = ({
price_data: {
product: config.stripe_product_id!,
unit_amount: Math.max(Math.round(amount * 100), 0),
unit_amount: Math.max(atmnToStripeAmount({ amount, currency }), 0),
currency,
},
period: {

4
server/src/external/stripe/stripeWebhookRouter.ts vendored
View File

@@ -17,11 +17,11 @@ export const stripeWebhookRouter = new Hono<StripeWebhookHonoEnv>();
stripeWebhookRouter.post(
"/webhooks/stripe/:orgId/:env",
stripeLegacySeederMiddleware,
stripeToAutumnCustomerMiddleware,
stripeIdempotencyMiddleware,
stripeWebhookEarlyAckMiddleware,
stripeWebhookRefreshMiddleware,
stripeSyncMiddleware,
stripeToAutumnCustomerMiddleware,
stripeLoggerMiddleware,
traceEnrichMiddleware,
handleStripeWebhookEvent,
@@ -31,11 +31,11 @@ stripeWebhookRouter.post(
stripeWebhookRouter.post(
"/webhooks/connect/:env",
stripeConnectSeederMiddleware,
stripeToAutumnCustomerMiddleware,
stripeIdempotencyMiddleware,
stripeWebhookEarlyAckMiddleware,
stripeWebhookRefreshMiddleware,
stripeSyncMiddleware,
stripeToAutumnCustomerMiddleware,
stripeLoggerMiddleware,
traceEnrichMiddleware,
handleStripeWebhookEvent,

28
server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/handleCheckoutSessionMetadataV2.ts vendored
View File

@@ -7,9 +7,9 @@ import { createStripeScheduleFromCheckout } from "@/external/stripe/webhookHandl
import { modifyStripeSubscriptionFromCheckout } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/modifyStripeSubscriptionFromCheckout";
import { syncSubscriptionItemMetadataFromCheckout } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/syncSubscriptionItemMetadataFromCheckout";
import { updateBillingPlanFromCheckout } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/updateBillingPlanFromCheckout";
import { withClaimedCheckoutSessionMetadata } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/withClaimedCheckoutSessionMetadata";
import type { StripeWebhookContext } from "@/external/stripe/webhookMiddlewares/stripeWebhookContext";
import { persistDeferredCreateSchedule } from "@/internal/billing/v2/actions/createSchedule/utils/persistDeferredCreateSchedule";
import { checkoutSessionLock } from "@/internal/billing/v2/actions/locks/checkoutSessionLock/checkoutSessionLock";
import { addStripeSubscriptionScheduleIdToBillingPlan } from "@/internal/billing/v2/execute/addStripeSubscriptionScheduleIdToBillingPlan";
import { executeAutumnBillingPlan } from "@/internal/billing/v2/execute/executeAutumnBillingPlan";
import { logAutumnBillingPlan } from "@/internal/billing/v2/utils/logs/logAutumnBillingPlan";
@@ -34,6 +34,24 @@ export const handleCheckoutSessionMetadataV2 = async ({
`[checkout.completed] Handling checkout session metadata V2: ${metadata.id}`,
);
await withClaimedCheckoutSessionMetadata({
ctx,
checkoutContext,
metadata,
execute: () =>
executeCheckoutSessionMetadataV2({ ctx, checkoutContext, metadata }),
});
};
const executeCheckoutSessionMetadataV2 = async ({
ctx,
checkoutContext,
metadata,
}: {
ctx: StripeWebhookContext;
checkoutContext: CheckoutSessionCompletedContext;
metadata: NonNullable<CheckoutSessionCompletedContext["metadata"]>;
}): Promise<void> => {
const deferredData = metadata.data as DeferredAutumnBillingPlanData;
// 1. Sync Autumn metadata onto subscription items created by checkout
@@ -96,14 +114,6 @@ export const handleCheckoutSessionMetadataV2 = async ({
billingPlan: updatedDeferredData.billingPlan,
});
// Clear checkout session lock now that customer_product rows exist
const lockCustomerId =
updatedDeferredData.billingContext.fullCustomer.id ??
updatedDeferredData.billingContext.fullCustomer.internal_id;
if (lockCustomerId) {
await checkoutSessionLock.clear({ ctx, customerId: lockCustomerId });
}
// Queue customer.products.updated webhook (mirrors executeBillingPlan)
await billingPlanToSendProductsUpdated({
ctx,

84
server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/withClaimedCheckoutSessionMetadata.ts vendored Normal file
View File

@@ -0,0 +1,84 @@
import {
type DeferredAutumnBillingPlanData,
MetadataType,
} from "@autumn/shared";
import { setStripeSubscriptionLock } from "@/external/stripe/subscriptions/utils/lockStripeSubscriptionUtils";
import type { CheckoutSessionCompletedContext } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/setupCheckoutSessionCompletedContext";
import type { StripeWebhookContext } from "@/external/stripe/webhookMiddlewares/stripeWebhookContext";
import { checkoutSessionLock } from "@/internal/billing/v2/actions/locks/checkoutSessionLock/checkoutSessionLock";
import { MetadataService } from "@/internal/metadata/MetadataService";
/**
* Runs `execute` exactly once across concurrent executors of the same deferred
* plan. The subscription lock marks the resulting subscription.updated events
* as Autumn-initiated; the checkout session lock is cleared even on failure
* since the Stripe session is already paid.
*/
export const withClaimedCheckoutSessionMetadata = async ({
ctx,
checkoutContext,
metadata,
execute,
}: {
ctx: StripeWebhookContext;
checkoutContext: CheckoutSessionCompletedContext;
metadata: NonNullable<CheckoutSessionCompletedContext["metadata"]>;
execute: () => Promise<void>;
}): Promise<void> => {
const claimed = await MetadataService.claim({
db: ctx.db,
id: metadata.id,
fromType: MetadataType.CheckoutSessionV2,
toType: MetadataType.CheckoutSessionV2Processing,
});
if (!claimed) {
ctx.logger.info(
`[checkout.completed] Metadata ${metadata.id} already claimed by another executor, skipping`,
);
return;
}
const deferredData = metadata.data as DeferredAutumnBillingPlanData;
const lockCustomerId =
deferredData?.billingContext?.fullCustomer?.id ??
deferredData?.billingContext?.fullCustomer?.internal_id;
try {
if (checkoutContext.stripeSubscription) {
await setStripeSubscriptionLock({
stripeSubscriptionId: checkoutContext.stripeSubscription.id,
lockedAtMs: Date.now(),
});
}
await execute();
} catch (error) {
await revertMetadataClaim({ ctx, metadataId: metadata.id });
throw error;
} finally {
if (lockCustomerId) {
await checkoutSessionLock.clear({ ctx, customerId: lockCustomerId });
}
}
};
const revertMetadataClaim = async ({
ctx,
metadataId,
}: {
ctx: StripeWebhookContext;
metadataId: string;
}): Promise<void> => {
await MetadataService.claim({
db: ctx.db,
id: metadataId,
fromType: MetadataType.CheckoutSessionV2Processing,
toType: MetadataType.CheckoutSessionV2,
}).catch((revertError) => {
ctx.logger.error(
`[checkout.completed] Failed to revert metadata claim for ${metadataId}`,
{ revertError },
);
});
};

15
server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionCanceled/handleStripeSubscriptionCanceled.ts vendored
View File

@@ -1,4 +1,10 @@
import { AttachScenario, cp, type FullCusProduct } from "@autumn/shared";
import {
AttachScenario,
cp,
type FullCusProduct,
notNullish,
} from "@autumn/shared";
import { msToSeconds } from "@shared/utils/common/unixUtils";
import { getStripeSubscriptionLock } from "@/external/stripe/subscriptions/utils/lockStripeSubscriptionUtils";
import type { StripeWebhookContext } from "@/external/stripe/webhookMiddlewares/stripeWebhookContext";
import { addProductsUpdatedWebhookTask } from "@/internal/analytics/handlers/handleProductsUpdated";
@@ -68,6 +74,13 @@ export const handleStripeSubscriptionCanceled = async ({
if (!isActiveRecurringAndOnSub) continue;
// attach-set ends_at, not an external cancellation
const endedAtMatchesCancelAt =
notNullish(customerProduct.ended_at) &&
notNullish(cancelsAtMs) &&
msToSeconds(customerProduct.ended_at!) === msToSeconds(cancelsAtMs!);
if (endedAtMatchesCancelAt) continue;
const updates = {
canceled_at: canceledAtMs ?? Date.now(),
canceled: true,

3
server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionRenewed/handleStripeSubscriptionRenewed.ts vendored
View File

@@ -77,6 +77,9 @@ export const handleStripeSubscriptionRenewed = async ({
if (!valid) continue;
// attach-set ends_at expiry, not a cancellation
if (!customerProduct.canceled && !customerProduct.canceled_at) continue;
// Clear cancellation fields
const updates = {
canceled_at: null,

15
server/src/external/stripe/webhookMiddlewares/stripeConnectSeederMiddleware.ts vendored
View File

@@ -1,6 +1,7 @@
import {
type AppEnv,
AuthType,
ErrCode,
type Feature,
type Organization,
} from "@autumn/shared";
@@ -11,6 +12,7 @@ import {
initMasterStripe,
} from "@/external/connect/initStripeCli.js";
import { OrgService } from "@/internal/orgs/OrgService.js";
import RecaseError from "@/utils/errorUtils.js";
import { createStripeCli } from "../../connect/createStripeCli.js";
import type {
StripeWebhookContext,
@@ -100,7 +102,18 @@ export const stripeConnectSeederMiddleware = async (
});
org = data.org;
features = data.features;
} catch {
} catch (error) {
// Only ack accounts genuinely not linked to an org; any other failure
// (e.g. DB outage) must 500 so Stripe retries instead of dropping the event.
const isOrgNotFound =
error instanceof RecaseError && error.code === ErrCode.OrgNotFound;
if (!isOrgNotFound) {
logger.error(
`Failed to resolve org for Stripe account ${accountId}, returning 500 for Stripe to retry: ${error}`,
);
return c.json({ error: "Failed to resolve org for Stripe webhook" }, 500);
}
if (process.env.NODE_ENV !== "development") {
logger.error(
`Account ID ${accountId} not linked to any org, skipping Stripe webhook`,

2
server/src/internal/analytics/actions/getCountAndSum.ts
View File

@@ -129,7 +129,7 @@ export const getCountAndSum = async ({
`
: `
SELECT event_name, sum(event_count) as count, sum(total_value) as sum
FROM ${useOrgRollup ? "events_org_hourly_mv" : "events_hourly_no_properties_two_mv"}
FROM ${useOrgRollup ? "events_org_hourly_mv" : "events_customer_hourly_mv"}
WHERE org_id = {org_id:String} AND env = {env:String}
${!useOrgRollup && !params.aggregateAll ? "AND customer_id = {customer_id:String}" : ""}
${!useOrgRollup && params.entity_id ? "AND entity_id = {entity_id:String}" : ""}

1
server/src/internal/balances/check/getCheckResponseV2.ts
View File

@@ -55,6 +55,7 @@ export const getCheckResponseV2 = async ({
apiSubject: evaluationApiSubject,
feature: featureToUse,
requiredBalance,
originalFeature,
}).allowed
: false;

2
server/src/internal/balances/check/runCheckWithRollout.ts
View File

@@ -3,6 +3,7 @@ import { withRedisFailOpen } from "@/external/redis/utils/withRedisFailOpen.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import type { CheckData } from "@/internal/api/check/checkTypes/CheckData.js";
import { getCheckFailOpenFallback } from "@/internal/api/check/checkUtils/getCheckFailOpenFallback.js";
import { isFullSubjectGateRejection } from "@/internal/customers/repos/getFullSubject/getFullSubjectGate.js";
import { isFullSubjectRolloutEnabled } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
import type { CheckDataV2 } from "./checkTypes/CheckDataV2.js";
import { runCheckLegacyFlow } from "./runCheckLegacyFlow.js";
@@ -37,6 +38,7 @@ export const runCheckWithRollout = async ({
return withRedisFailOpen<RunCheckResult<CheckData | CheckDataV2>>({
source: "runCheckWithRollout",
run: () => runCheckV2({ ctx, body, requiredBalance }),
alsoFailOpen: isFullSubjectGateRejection,
fallback: (error) => ({
checkData: null,
response: getCheckFailOpenFallback({

6
server/src/internal/balances/check/runCheckWithTrackV2.ts
View File

@@ -10,6 +10,7 @@ import {
type ParsedCheckParams,
RecaseError,
type TrackParams,
UsageLimitExceededError,
} from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { getTrackFeatureDeductions } from "@/internal/balances/track/utils/getFeatureDeductions.js";
@@ -94,7 +95,10 @@ export const runCheckWithTrackV2 = async ({
checkData.evaluationApiBalance = trackedBalance ?? undefined;
trackBalances = response.balances;
} catch (error) {
if (error instanceof InsufficientBalanceError) {
if (
error instanceof InsufficientBalanceError ||
error instanceof UsageLimitExceededError
) {
allowed = false;
} else {
throw error;

16
server/src/internal/balances/finalizeLock/runRedisFinalizeLockV2.ts
View File

@@ -41,13 +41,20 @@ export const runRedisFinalizeLockV2 = async ({
throw error;
}
const { updates, rolloverUpdates, modifiedCusEntIdsByFeatureId } =
redisResult;
const {
updates,
rolloverUpdates,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
} = redisResult;
const modifiedCusEntIds = deductionUpdatesToModifiedIds({ updates });
const rolloverIds = Object.keys(rolloverUpdates);
if (modifiedCusEntIds.length > 0 || rolloverIds.length > 0) {
if (
modifiedCusEntIds.length > 0 ||
rolloverIds.length > 0 ||
usageWindowUpdates.length > 0
) {
globalSyncBatchingManagerV3.addSyncItem({
customerId: receipt.customer_id,
orgId: ctx.org.id,
@@ -57,6 +64,7 @@ export const runRedisFinalizeLockV2 = async ({
region: currentRegion,
entityId: receipt.entity_id ?? undefined,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
});
}

2
server/src/internal/balances/track/runTrackWithRollout.ts
View File

@@ -1,6 +1,7 @@
import type { ApiVersion, TrackParams, TrackResponseV3 } from "@autumn/shared";
import { withRedisFailOpen } from "@/external/redis/utils/withRedisFailOpen.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { isFullSubjectGateRejection } from "@/internal/customers/repos/getFullSubject/getFullSubjectGate.js";
import { isFullSubjectRolloutEnabled } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
import type { FeatureDeduction } from "../utils/types/featureDeduction.js";
import { runTrackV2 } from "./runTrackV2.js";
@@ -38,6 +39,7 @@ export const runTrackWithRollout = async ({
featureDeductions,
apiVersion,
}),
alsoFailOpen: isFullSubjectGateRejection,
fallback: async (error) => {
const queuedResponse = await queueTrack({ ctx, body });
if (queuedResponse) return queuedResponse;

2
server/src/internal/balances/track/v3/handleRedisTrackErrorV3.ts
View File

@@ -6,8 +6,8 @@ import {
type TrackParams,
type TrackResponseV3,
} from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { RedisUnavailableError } from "@/external/redis/utils/errors.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import type { FeatureDeduction } from "../../utils/types/featureDeduction.js";
import {
RedisDeductionError,

17
server/src/internal/balances/track/v3/runRedisTrackV3.ts
View File

@@ -7,11 +7,11 @@ import type {
import { tryCatch } from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { globalEventBatchingManager } from "@/internal/balances/events/EventBatchingManager.js";
import { resolveInternalProductIdForEvent } from "@/internal/balances/events/resolveInternalProductIdForEvent.js";
import {
buildEventInfo,
initEvent,
} from "@/internal/balances/events/initEvent.js";
import { resolveInternalProductIdForEvent } from "@/internal/balances/events/resolveInternalProductIdForEvent.js";
import {
deductionToTrackResponseV2,
executeRedisDeductionV2,
@@ -20,6 +20,7 @@ import {
import { globalSyncBatchingManagerV3 } from "@/internal/balances/utils/sync/SyncBatchingManagerV3.js";
import type { FeatureDeduction } from "../../utils/types/featureDeduction.js";
import type { RolloverUpdate } from "../../utils/types/rolloverUpdate.js";
import type { UsageWindowUpdate } from "../../utils/types/usageWindowUpdate.js";
import { buildAiCreditCostProperty } from "../utils/buildAiCreditCostProperty.js";
import { handleRedisTrackErrorV3 } from "./handleRedisTrackErrorV3.js";
@@ -29,18 +30,25 @@ const queueSyncItem = ({
fullSubject,
rolloverUpdates,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
}: {
ctx: AutumnContext;
body: TrackParams;
fullSubject: FullSubject;
rolloverUpdates: Record<string, RolloverUpdate>;
modifiedCusEntIdsByFeatureId: Record<string, string[]>;
usageWindowUpdates?: UsageWindowUpdate[];
}): void => {
const cusEntIds = Object.values(modifiedCusEntIdsByFeatureId).flat();
const rolloverIds = Object.keys(rolloverUpdates);
if (cusEntIds.length === 0 && rolloverIds.length === 0) return;
if (
cusEntIds.length === 0 &&
rolloverIds.length === 0 &&
(usageWindowUpdates?.length ?? 0) === 0
) {
return;
}
globalSyncBatchingManagerV3.addSyncItem({
customerId: body.customer_id,
@@ -50,6 +58,7 @@ const queueSyncItem = ({
rolloverIds,
entityId: fullSubject.entityId,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
});
};
@@ -129,6 +138,7 @@ export const runRedisTrackV3 = async ({
rolloverUpdates,
modifiedCusEntIdsByFeatureId,
mutationLogs,
usageWindowUpdates,
} = result;
queueSyncItem({
@@ -137,6 +147,7 @@ export const runRedisTrackV3 = async ({
fullSubject: updatedFullSubject,
rolloverUpdates,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
});
const deductions = projectMutationLogsToTrackDeductionsV2({

10
server/src/internal/balances/updateBalance/v2/updateRemainingV2.ts
View File

@@ -67,11 +67,16 @@ export const updateRemainingV2 = async ({
});
}
const { rolloverUpdates, modifiedCusEntIdsByFeatureId } = result;
const { rolloverUpdates, modifiedCusEntIdsByFeatureId, usageWindowUpdates } =
result;
const cusEntIds = Object.values(modifiedCusEntIdsByFeatureId).flat();
const rolloverIds = Object.keys(rolloverUpdates);
if (cusEntIds.length > 0 || rolloverIds.length > 0) {
if (
cusEntIds.length > 0 ||
rolloverIds.length > 0 ||
usageWindowUpdates.length > 0
) {
await syncItemV4({
ctx,
payload: {
@@ -82,6 +87,7 @@ export const updateRemainingV2 = async ({
rolloverIds,
entityId: fullSubject.entityId,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
},
});
}

10
server/src/internal/balances/updateBalance/v2/updateUsageV2.ts
View File

@@ -111,11 +111,16 @@ export const updateUsageV2 = async ({
});
}
const { rolloverUpdates, modifiedCusEntIdsByFeatureId } = result;
const { rolloverUpdates, modifiedCusEntIdsByFeatureId, usageWindowUpdates } =
result;
const cusEntIds = Object.values(modifiedCusEntIdsByFeatureId).flat();
const rolloverIds = Object.keys(rolloverUpdates);
if (cusEntIds.length > 0 || rolloverIds.length > 0) {
if (
cusEntIds.length > 0 ||
rolloverIds.length > 0 ||
usageWindowUpdates.length > 0
) {
await syncItemV4({
ctx,
payload: {
@@ -126,6 +131,7 @@ export const updateUsageV2 = async ({
rolloverIds,
entityId: fullSubject.entityId,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
},
});
}

36
server/src/internal/balances/utils/deductionV2/applyUsageWindowUpdatesToFullSubject.ts Normal file
View File

@@ -0,0 +1,36 @@
import type { FullSubject, UsageWindow } from "@autumn/shared";
/**
* Refresh the in-flight subject's customer-scoped usage-window counters from
* the deduction result, so usage_limit_used (webhooks, API responses built
* from this subject) reflects the deduction. Sibling of
* applyDeductionUpdateToFullSubject / applyRolloverUpdatesToFullSubject.
*
* The Lua result carries ALL scopes; the subject keeps its own scope only
* (entity subjects hold just their entity's rows).
*/
export const applyUsageWindowUpdatesToFullSubject = ({
fullSubject,
usageWindowsByFeatureId,
}: {
fullSubject: FullSubject;
usageWindowsByFeatureId: Record<string, UsageWindow[]> | null | undefined;
}): void => {
if (!usageWindowsByFeatureId) return;
const updatedFeatureIds = new Set(Object.keys(usageWindowsByFeatureId));
const updatedWindows = Object.values(usageWindowsByFeatureId)
.flat()
.filter((usageWindow) =>
fullSubject.internalEntityId
? usageWindow.internal_entity_id === fullSubject.internalEntityId
: true,
);
fullSubject.usage_windows = [
...(fullSubject.usage_windows ?? []).filter(
(usageWindow) => !updatedFeatureIds.has(usageWindow.feature_id),
),
...updatedWindows,
];
};

4
server/src/internal/balances/utils/deductionV2/executePostgresDeductionV2.ts
View File

@@ -117,6 +117,7 @@ export const executePostgresDeductionV2 = async ({
fullSubject,
deduction,
options: resolvedOptions,
now: Date.now(),
});
if (customerEntitlements.length === 0 || unlimitedFeatureIds.length > 0) {
@@ -147,6 +148,9 @@ export const executePostgresDeductionV2 = async ({
sql`SELECT * FROM deduct_from_cus_ents(
${JSON.stringify({
sorted_entitlements: customerEntitlementDeductions,
// No usage_window_limits here: the hard usage cap is enforced only on the
// Redis/Lua path, so this Postgres fallback intentionally fails open
// (availability over strict cap enforcement during a Redis outage).
spend_limit_by_feature_id: spendLimitByFeatureId ?? null,
usage_based_cus_ent_ids_by_feature_id:
usageBasedCusEntIdsByFeatureId ?? null,

59
server/src/internal/balances/utils/deductionV2/executeRedisDeductionV2.ts
View File

@@ -2,6 +2,7 @@ import {
type FullCusEntWithFullCusProduct,
type FullSubject,
fullSubjectToFullCustomer,
notNullish,
} from "@autumn/shared";
import type { Redis } from "ioredis";
import { currentRegion } from "@/external/redis/initRedis.js";
@@ -16,6 +17,7 @@ import { createAllocatedInvoice } from "@/internal/balances/utils/allocatedInvoi
import { saveLockReceiptV2 } from "@/internal/balances/utils/lockV2/saveLockReceiptV2.js";
import { buildDeductFromSubjectBalancesKeys } from "@/internal/customers/cache/fullSubject/builders/buildDeductFromSubjectBalancesKeys.js";
import { buildFullSubjectKey } from "@/internal/customers/cache/fullSubject/builders/buildFullSubjectKey.js";
import { FULL_SUBJECT_CACHE_TTL_SECONDS } from "@/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.js";
import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
import type { DeductionOptions } from "../types/deductionTypes.js";
import type { DeductionUpdate } from "../types/deductionUpdate.js";
@@ -27,8 +29,11 @@ import {
} from "../types/redisDeductionError.js";
import type { LuaDeductionResult } from "../types/redisDeductionResult.js";
import type { RolloverUpdate } from "../types/rolloverUpdate.js";
import type { UsageWindowMutation } from "../types/usageWindowMutation.js";
import type { UsageWindowUpdate } from "../types/usageWindowUpdate.js";
import { applyDeductionUpdateToFullSubject } from "./applyDeductionUpdateToFullSubject.js";
import { applyRolloverUpdatesToFullSubject } from "./applyRolloverUpdatesToFullSubject.js";
import { applyUsageWindowUpdatesToFullSubject } from "./applyUsageWindowUpdatesToFullSubject.js";
import { buildUnlimitedPlanMutationLog } from "./buildUnlimitedPlanMutationLog.js";
import { logDeductionUpdatesV2 } from "./logDeductionUpdatesV2.js";
import { mutationLogsToFeaturesV2 } from "./mutationLogsToFeaturesV2.js";
@@ -60,6 +65,8 @@ export const executeRedisDeductionV2 = async ({
rolloverUpdates: Record<string, RolloverUpdate>;
mutationLogs: MutationLogItem[];
modifiedCusEntIdsByFeatureId: Record<string, string[]>;
usageWindowUpdates: UsageWindowUpdate[];
usageWindowMutations: UsageWindowMutation[];
}> => {
const { org, env } = ctx;
const oldFullSubject = structuredClone(fullSubject);
@@ -95,7 +102,11 @@ export const executeRedisDeductionV2 = async ({
let allUpdates: Record<string, DeductionUpdate> = {};
let allRolloverUpdates: Record<string, RolloverUpdate> = {};
let allMutationLogs: MutationLogItem[] = [];
let allUsageWindowMutations: UsageWindowMutation[] = [];
const allModifiedCusEntIdsByFeatureId: Record<string, string[]> = {};
// Keyed by feature id: each Lua result carries the COMPLETE post-deduction
// counter array per capped feature, so last write wins across deductions.
const allUsageWindowUpdates: Record<string, UsageWindowUpdate> = {};
const customerId = fullSubject.customerId;
const routingKey = buildFullSubjectKey({
@@ -105,6 +116,10 @@ export const executeRedisDeductionV2 = async ({
entityId: fullSubject.entityId,
});
// One timestamp for the whole operation: the resolver keys windows from it
// and Lua receives the same value, so they never disagree on the window.
const usageWindowNow = Date.now();
for (const deduction of deductions) {
const {
feature,
@@ -118,6 +133,8 @@ export const executeRedisDeductionV2 = async ({
customerEntitlementDeductions,
spendLimitByFeatureId,
usageBasedCusEntIdsByFeatureId,
usageWindowLimits,
usageWindowFeatureIds,
rollovers,
customerEntitlements,
unlimitedFeatureIds,
@@ -128,6 +145,7 @@ export const executeRedisDeductionV2 = async ({
fullSubject,
deduction,
options,
now: usageWindowNow,
});
if (unlimitedFeatureIds.length > 0) {
@@ -172,8 +190,17 @@ export const executeRedisDeductionV2 = async ({
idempotencyKey: idempotencyRedisKey,
customerEntitlementDeductions,
fallbackFeatureId: feature.id,
usageWindowFeatureIds,
});
// Usage windows are enforced/incremented only for real positive
// consumption, never for target_balance set-downs or granted-balance edits.
const isConsumption =
notNullish(toDeduct) &&
(toDeduct as number) > 0 &&
!notNullish(targetBalance) &&
!options.alterGrantedBalance;
const luaParams = {
org_id: org.id,
env,
@@ -183,6 +210,10 @@ export const executeRedisDeductionV2 = async ({
spend_limit_by_feature_id: spendLimitByFeatureId ?? null,
usage_based_cus_ent_ids_by_feature_id:
usageBasedCusEntIdsByFeatureId ?? null,
usage_window_limits: usageWindowLimits ?? null,
usage_window_now: usageWindowNow,
usage_window_ttl_seconds: FULL_SUBJECT_CACHE_TTL_SECONDS,
is_consumption: isConsumption,
amount_to_deduct: toDeduct ?? null,
target_balance: targetBalance ?? null,
target_entity_id: entityId || null,
@@ -234,6 +265,7 @@ export const executeRedisDeductionV2 = async ({
throw new RedisDeductionError({
message: `Redis deduction failed: ${resultJson.error}`,
code: resultJson.error as RedisDeductionErrorCode,
featureId: resultJson.feature_id,
});
}
@@ -241,6 +273,11 @@ export const executeRedisDeductionV2 = async ({
const mutationLogs = Array.isArray(resultJson.mutation_logs)
? resultJson.mutation_logs
: [];
const usageWindowMutations = Array.isArray(
resultJson.usage_window_mutations,
)
? resultJson.usage_window_mutations
: [];
const modifiedCustomerEntitlementIds = Array.isArray(
resultJson.modified_customer_entitlement_ids,
)
@@ -257,6 +294,21 @@ export const executeRedisDeductionV2 = async ({
allUpdates = { ...allUpdates, ...updates };
allRolloverUpdates = { ...allRolloverUpdates, ...rollover_updates };
allMutationLogs = [...allMutationLogs, ...mutationLogs];
allUsageWindowMutations = [
...allUsageWindowMutations,
...usageWindowMutations,
];
// Typed handoff for the PG mirror; empty arrays kept (prune-to-empty
// must still full-replace).
for (const [featureId, usageWindows] of Object.entries(
resultJson.usage_windows_by_feature_id ?? {},
)) {
allUsageWindowUpdates[featureId] = {
internal_customer_id: fullSubject.internalCustomerId,
feature_id: featureId,
usage_windows: usageWindows,
};
}
const syncState = normalizeDeductionSyncStateV2({
customerEntitlements,
@@ -281,6 +333,11 @@ export const executeRedisDeductionV2 = async ({
rolloverUpdates: rollover_updates,
});
applyUsageWindowUpdatesToFullSubject({
fullSubject,
usageWindowsByFeatureId: resultJson.usage_windows_by_feature_id,
});
for (const customerEntitlementId of Object.keys(updates)) {
const update = updates[customerEntitlementId];
const customerEntitlement = customerEntitlements.find(
@@ -355,5 +412,7 @@ export const executeRedisDeductionV2 = async ({
rolloverUpdates: allRolloverUpdates,
mutationLogs: allMutationLogs,
modifiedCusEntIdsByFeatureId: allModifiedCusEntIdsByFeatureId,
usageWindowUpdates: Object.values(allUsageWindowUpdates),
usageWindowMutations: allUsageWindowMutations,
};
};

49
server/src/internal/balances/utils/deductionV2/prepareFeatureDeductionV2.ts
View File

@@ -1,22 +1,26 @@
import {
AllowanceType,
cusEntToStartingBalance,
ErrCode,
type FullCusEntWithFullCusProduct,
type FullSubject,
fullSubjectToCustomerEntitlements,
fullSubjectToOverageAllowedByFeatureId,
fullSubjectToSpendLimitByFeatureId,
fullSubjectToUsageBasedCusEntsByFeatureId,
fullSubjectToUsageWindowLimits,
getMaxOverage,
getRelevantFeatures,
isAllocatedCustomerEntitlement,
isFreeCustomerEntitlement,
notNullish,
orgToInStatuses,
RecaseError,
} from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { buildLockReceiptKey } from "@/internal/balances/utils/lock/buildLockReceiptKey.js";
import { getUnlimitedAndUsageAllowed } from "@/internal/customers/cusProducts/cusEnts/cusEntUtils.js";
import { generateId } from "@/utils/genUtils.js";
import { computeCreditCosts } from "../deduction/computeCreditCosts.js";
import type {
CustomerEntitlementDeduction,
@@ -34,11 +38,15 @@ export const prepareFeatureDeductionV2 = ({
fullSubject,
deduction,
options = {},
now,
}: {
ctx: AutumnContext;
fullSubject: FullSubject;
deduction: FeatureDeduction;
options?: DeductionOptions;
// Single timestamp shared with the Lua param so the resolved window key and
// the script agree on which window a boundary-crossing request lands in.
now: number;
}): PreparedFeatureDeduction => {
const { org, env } = ctx;
const { feature, lock, targetBalance } = deduction;
@@ -108,6 +116,41 @@ export const prepareFeatureDeductionV2 = ({
fullSubject,
featureIds: effectiveFeatureIds,
});
// Resolve windows against the full relevant set (incl credit-system parents)
// even under set_usage, so a parent-feature cap can't be bypassed by set_usage
// on a member feature.
const windowFeatureIds = notNullish(targetBalance)
? getRelevantFeatures({
features: ctx.features,
featureId: feature.id,
}).map((candidate) => candidate.id)
: effectiveFeatureIds;
const usageWindowLimits = fullSubjectToUsageWindowLimits({
fullSubject,
featureIds: windowFeatureIds,
features: ctx.features,
now,
inStatuses: orgToInStatuses({ org }),
});
// Counters are customer-scoped: a null anchor only means calendar-aligned
// bounds with no provenance, not an unenforceable cap.
for (const windowLimit of usageWindowLimits) {
windowLimit.new_window_id = generateId("uw");
if (windowLimit.anchor_customer_entitlement_id === null) {
ctx.logger.warn(
`usage window for feature ${windowLimit.feature_id} has no anchor entitlement; using calendar-aligned bounds with no provenance.`,
);
}
}
// set_usage carries no window provenance, so it would silently bypass the hard
// cap; reject it when the feature has an enforced usage window.
if (notNullish(targetBalance) && usageWindowLimits.length > 0) {
throw new RecaseError({
message: `Cannot set usage for feature ${feature.id}: it has an active usage limit. Remove or adjust the limit, or record usage normally instead of using set_usage.`,
code: ErrCode.SetUsageNotAllowedWithUsageLimit,
});
}
const nativeUsageAllowedFeatureIds = new Set(
customerEntitlements
@@ -211,6 +254,12 @@ export const prepareFeatureDeductionV2 = ({
Object.keys(usageBasedCusEntIdsByFeatureId).length > 0
? usageBasedCusEntIdsByFeatureId
: undefined,
usageWindowLimits:
usageWindowLimits.length > 0 ? usageWindowLimits : undefined,
usageWindowFeatureIds:
usageWindowLimits.length > 0
? [...new Set(usageWindowLimits.map((limit) => limit.feature_id))]
: undefined,
rollovers: sortedRollovers.map((rollover) => ({
id: rollover.id,
credit_cost: rollover.credit_cost,

83
server/src/internal/balances/utils/sql/syncBalancesV2.sql
View File

@@ -14,6 +14,12 @@
-- - balance: number
-- - usage: number
-- - entities: jsonb (the full entities object)
-- usage_window_updates: array of objects with:
-- - internal_customer_id: string
-- - feature_id: string
-- - usage_windows: jsonb array of DbUsageWindow rows (the COMPLETE set for
-- that customer+feature; Redis is authoritative and prunes closed
-- windows, so rows are full-replaced per customer+feature)
--
-- Returns JSONB with:
-- updates: object mapping customer_entitlement_id -> { balance, adjustment, entities }
@@ -32,6 +38,7 @@ AS $$
DECLARE
customer_entitlement_updates jsonb := params->'customer_entitlement_updates';
rollover_updates_param jsonb := params->'rollover_updates';
usage_window_updates_param jsonb := params->'usage_window_updates';
ent_obj jsonb;
ent_id text;
@@ -42,6 +49,11 @@ DECLARE
ent_entity_count int;
ent_cache_version int;
uw_obj jsonb;
uw_internal_customer_id text;
uw_feature_id text;
uw_windows jsonb;
db_next_reset_at bigint;
db_entity_count int;
db_cache_version int;
@@ -134,7 +146,6 @@ BEGIN
ent_id, ent_cache_version, db_cache_version;
END IF;
-- Update the customer_entitlement row directly
UPDATE customer_entitlements ce
SET
balance = COALESCE(ent_balance, ce.balance),
@@ -142,7 +153,6 @@ BEGIN
entities = COALESCE(ent_entities, ce.entities)
WHERE ce.id = ent_id;
-- Track update
IF FOUND THEN
updates_json := jsonb_set(
updates_json,
@@ -191,6 +201,75 @@ BEGIN
END LOOP;
END IF;
-- ============================================================================
-- STEP 4: Mirror usage-window counters (race-safe upsert)
-- ============================================================================
-- ONE mutable row per (customer, feature, entity scope); bounds roll in
-- place. Upsert on the scope key (never on id) so concurrent creates can't
-- abort, with an updated_at guard so older snapshots never clobber newer.
IF usage_window_updates_param IS NOT NULL THEN
FOR uw_obj IN SELECT * FROM jsonb_array_elements(usage_window_updates_param)
LOOP
uw_internal_customer_id := uw_obj->>'internal_customer_id';
uw_feature_id := uw_obj->>'feature_id';
uw_windows := uw_obj->'usage_windows';
IF uw_internal_customer_id IS NOT NULL
AND uw_feature_id IS NOT NULL
AND uw_windows IS NOT NULL
AND uw_windows != 'null'::jsonb THEN
IF jsonb_typeof(uw_windows) != 'array' THEN
uw_windows := '[]'::jsonb;
END IF;
INSERT INTO usage_windows (
id, internal_customer_id, internal_entity_id, feature_id,
internal_feature_id, anchor_customer_entitlement_id,
window_start_at, window_end_at, usage, updated_at
)
SELECT
w->>'id',
uw_internal_customer_id,
w->>'internal_entity_id',
uw_feature_id,
w->>'internal_feature_id',
CASE
WHEN w->>'anchor_customer_entitlement_id' IS NOT NULL
AND EXISTS (
SELECT 1 FROM customer_entitlements ce
WHERE ce.id = w->>'anchor_customer_entitlement_id'
)
THEN w->>'anchor_customer_entitlement_id'
ELSE NULL
END,
(w->>'window_start_at')::numeric,
(w->>'window_end_at')::numeric,
(w->>'usage')::numeric,
(w->>'updated_at')::numeric
FROM jsonb_array_elements(uw_windows) AS w
WHERE w->>'id' IS NOT NULL
AND w->>'internal_feature_id' IS NOT NULL
AND EXISTS (
SELECT 1 FROM features f
WHERE f.internal_id = w->>'internal_feature_id'
)
ON CONFLICT (
internal_customer_id, internal_feature_id,
COALESCE(internal_entity_id, '')
)
DO UPDATE SET
usage = EXCLUDED.usage,
updated_at = EXCLUDED.updated_at,
window_start_at = EXCLUDED.window_start_at,
window_end_at = EXCLUDED.window_end_at,
feature_id = EXCLUDED.feature_id,
anchor_customer_entitlement_id =
EXCLUDED.anchor_customer_entitlement_id
WHERE EXCLUDED.updated_at >= usage_windows.updated_at;
END IF;
END LOOP;
END IF;
RETURN jsonb_build_object(
'updates', updates_json,
'rollover_updates', rollover_updates_json

36
server/src/internal/balances/utils/sync/SyncBatchingManagerV3.ts
View File

@@ -3,6 +3,7 @@ import { logger } from "@/external/logtail/logtailUtils.js";
import { currentRegion } from "@/external/redis/initRedis.js";
import { JobName } from "@/queue/JobName.js";
import { addTaskToQueue } from "@/queue/queueUtils.js";
import type { UsageWindowUpdate } from "../types/usageWindowUpdate.js";
interface CustomerBatchContext {
customerId: string;
@@ -14,6 +15,10 @@ interface CustomerBatchContext {
rolloverIds: Set<string>;
entityId?: string;
modifiedCusEntIdsByFeatureId: Record<string, string[]>;
// Counter SNAPSHOTS keyed by capped feature: each deduction returns the
// complete post-deduction array, so merging across batched items is
// last-write-wins (unlike cusEnt/rollover ids, which accumulate).
usageWindowUpdatesByFeatureId: Record<string, UsageWindowUpdate>;
}
interface CustomerBatch {
@@ -33,6 +38,7 @@ export type QueueSyncV4Payload = {
rolloverIds: string[];
entityId?: string;
modifiedCusEntIdsByFeatureId: Record<string, string[]>;
usageWindowUpdates?: UsageWindowUpdate[];
};
messageGroupId?: string;
messageDeduplicationId: string;
@@ -78,6 +84,7 @@ export class SyncBatchingManagerV3 {
region,
entityId,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
}: {
customerId: string;
orgId: string;
@@ -87,6 +94,7 @@ export class SyncBatchingManagerV3 {
region?: string;
entityId?: string;
modifiedCusEntIdsByFeatureId: Record<string, string[]>;
usageWindowUpdates?: UsageWindowUpdate[];
}): void {
const batchKey = this.buildBatchKey({ orgId, env, customerId });
let batch = this.customerBatches.get(batchKey);
@@ -112,6 +120,12 @@ export class SyncBatchingManagerV3 {
batch.context.modifiedCusEntIdsByFeatureId[featureId].push(...ids);
}
for (const usageWindowUpdate of usageWindowUpdates ?? []) {
batch.context.usageWindowUpdatesByFeatureId[
usageWindowUpdate.feature_id
] = usageWindowUpdate;
}
const totalSize =
batch.context.cusEntIds.size + batch.context.rolloverIds.size;
if (totalSize >= this.MAX_BATCH_SIZE) {
@@ -177,6 +191,7 @@ export class SyncBatchingManagerV3 {
cusEntIds: new Set(),
rolloverIds: new Set(),
modifiedCusEntIdsByFeatureId: {},
usageWindowUpdatesByFeatureId: {},
},
timer: null,
};
@@ -231,7 +246,13 @@ export class SyncBatchingManagerV3 {
this.customerBatches.delete(batchKey);
const { context } = batch;
if (context.cusEntIds.size === 0 && context.rolloverIds.size === 0) return;
if (
context.cusEntIds.size === 0 &&
context.rolloverIds.size === 0 &&
Object.keys(context.usageWindowUpdatesByFeatureId).length === 0
) {
return;
}
await this.queueSyncJob({ context });
}
@@ -247,10 +268,12 @@ export class SyncBatchingManagerV3 {
context,
cusEntIds,
rolloverIds,
usageWindowUpdates,
}: {
context: CustomerBatchContext;
cusEntIds: string[];
rolloverIds: string[];
usageWindowUpdates: UsageWindowUpdate[];
}): string {
const dedupBucket = Math.floor(Date.now() / this.DEDUP_BUCKET_MS);
const dedupKey = JSON.stringify({
@@ -260,6 +283,10 @@ export class SyncBatchingManagerV3 {
customerId: context.customerId,
cusEntIds,
rolloverIds,
// Snapshots ride the payload (cusEnt balances are re-read at consume
// time, counters are not), so a newer snapshot must never be dropped
// as a duplicate of an older one within the bucket.
usageWindowUpdates,
dedupBucket,
});
@@ -273,10 +300,14 @@ export class SyncBatchingManagerV3 {
}): Promise<void> {
const cusEntIds = Array.from(context.cusEntIds).sort();
const rolloverIds = Array.from(context.rolloverIds).sort();
const usageWindowUpdates = Object.values(
context.usageWindowUpdatesByFeatureId,
);
const messageDeduplicationId = this.buildDeduplicationId({
context,
cusEntIds,
rolloverIds,
usageWindowUpdates,
});
try {
@@ -292,13 +323,14 @@ export class SyncBatchingManagerV3 {
rolloverIds,
entityId: context.entityId,
modifiedCusEntIdsByFeatureId: context.modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
},
// messageGroupId: `sync-v4:${context.orgId}:${context.env}:${context.customerId}`,
messageDeduplicationId,
});
logger.debug(
`[SyncV4] Queued sync for ${context.customerId}, ${cusEntIds.length} entitlements, ${rolloverIds.length} rollovers`,
`[SyncV4] Queued sync for ${context.customerId}, ${cusEntIds.length} entitlements, ${rolloverIds.length} rollovers, ${usageWindowUpdates.length} usage windows`,
);
} catch (error) {
logger.error(

37
server/src/internal/balances/utils/sync/syncItemV4.ts
View File

@@ -10,6 +10,7 @@ import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { getCachedFeatureBalance } from "@/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.js";
import { deleteCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.js";
import { globalRefreshEntityAggregateBatchingManager } from "../refreshEntityAggregate/RefreshEntityAggregateBatchingManager";
import type { UsageWindowUpdate } from "../types/usageWindowUpdate.js";
import { logSyncItem } from "./logs/logSyncItem";
const SYNC_CONFLICT_CODES = {
@@ -68,6 +69,10 @@ interface SyncItemV4 {
timestamp: number;
rolloverIds?: string[];
modifiedCusEntIdsByFeatureId: Record<string, string[]>;
/** Post-deduction counter snapshots handed straight from the Lua result
* (no Redis re-read); mirrored to the customer-scoped usage_windows table
* via full-replace per (customer, feature). */
usageWindowUpdates?: UsageWindowUpdate[];
}
export interface SyncEntry {
@@ -113,12 +118,17 @@ export const syncItemV4 = async ({
ctx: AutumnContext;
payload: SyncItemV4;
}): Promise<void> => {
const { customerId, entityId, rolloverIds, modifiedCusEntIdsByFeatureId } =
payload;
const {
customerId,
entityId,
rolloverIds,
modifiedCusEntIdsByFeatureId,
usageWindowUpdates,
} = payload;
const { db } = ctx;
// Read targeted balance hashes
const allSubjectBalances: SubjectBalance[] = [];
let allSubjectBalances: SubjectBalance[] = [];
for (const [featureId, customerEntitlementIds] of Object.entries(
modifiedCusEntIdsByFeatureId,
)) {
@@ -131,6 +141,9 @@ export const syncItemV4 = async ({
});
if (outcome.kind !== "ok") {
ctx.logger.warn(
`[SYNC V4] (${customerId}) Cache miss for feature ${featureId}; skipping this feature only.`,
);
logSyncItem({
ctx,
result: {
@@ -139,7 +152,11 @@ export const syncItemV4 = async ({
feature: featureId,
},
});
return;
// A miss (e.g. an invalidation racing the batch) drops the BALANCE
// sync wholesale, but usage-window snapshots ride in the payload and
// need no cache read -- they must still land.
allSubjectBalances = [];
break;
}
allSubjectBalances.push(...outcome.value.balances);
@@ -169,7 +186,16 @@ export const syncItemV4 = async ({
}
}
if (entries.length === 0 && rolloverEntries.length === 0) {
// Customer-scoped usage-window counters arrive pre-built from the deduction
// result (same atomic Lua execution that incremented them) -- no Redis
// re-read here. Full-replaced per (customer, feature) by the SQL function.
const usageWindowEntries: UsageWindowUpdate[] = usageWindowUpdates ?? [];
if (
entries.length === 0 &&
rolloverEntries.length === 0 &&
usageWindowEntries.length === 0
) {
logSyncItem({ ctx, result: { kind: "skipped", reason: "no_entries" } });
return;
}
@@ -179,6 +205,7 @@ export const syncItemV4 = async ({
sql`SELECT * FROM sync_balances_v2(${JSON.stringify({
customer_entitlement_updates: entries,
rollover_updates: rolloverEntries,
usage_window_updates: usageWindowEntries,
})}::jsonb)`,
),
);

7
server/src/internal/balances/utils/types/deductionTypes.ts
View File

@@ -2,6 +2,7 @@ import type {
CustomerEntitlementFilters,
DbSpendLimit,
FullCusEntWithFullCusProduct,
UsageWindowLimit,
} from "@autumn/shared";
/** Behavior options for deduction */
@@ -42,6 +43,12 @@ export type PreparedFeatureDeduction = {
customerEntitlementDeductions: CustomerEntitlementDeduction[];
spendLimitByFeatureId?: Record<string, DbSpendLimit>;
usageBasedCusEntIdsByFeatureId?: Record<string, string[]>;
// Resolved windowed usage-limit caps, enforced inside the deduction script.
usageWindowLimits?: UsageWindowLimit[];
// Distinct capped feature ids: their balance hashes carry the
// `_usage_windows` counter field, so their keys must be declared in KEYS[]
// even when no deduction entry references them.
usageWindowFeatureIds?: string[];
// rolloverIds: string[];
rollovers: RolloverDeduction[];
unlimitedFeatureIds: string[];

4
server/src/internal/balances/utils/types/redisDeductionError.ts
View File

@@ -23,17 +23,21 @@ export const FALLBACK_ERROR_CODES = [
/** Error thrown by Redis deduction operations */
export class RedisDeductionError extends Error {
code: RedisDeductionErrorCode;
featureId?: string;
constructor({
message,
code,
featureId,
}: {
message: string;
code: RedisDeductionErrorCode;
featureId?: string;
}) {
super(message);
this.name = "RedisDeductionError";
this.code = code;
this.featureId = featureId;
}
isRedisUnavailable(): boolean {

9
server/src/internal/balances/utils/types/redisDeductionResult.ts
View File

@@ -1,12 +1,21 @@
import type { UsageWindow } from "@autumn/shared";
import type { DeductionUpdate } from "./deductionUpdate.js";
import type { MutationLogItem } from "./mutationLogItem.js";
import type { RolloverUpdate } from "./rolloverUpdate.js";
import type { UsageWindowMutation } from "./usageWindowMutation.js";
export interface LuaDeductionResult {
updates: Record<string, DeductionUpdate>;
rollover_updates: Record<string, RolloverUpdate>;
modified_customer_entitlement_ids: string[];
mutation_logs: MutationLogItem[];
/** Post-deduction COUNTER ROWS per capped feature (usage amounts; mirrors
* the usage_windows table) -- not the limits config, which goes IN via
* usage_window_limits. Null when no usage windows were enforced. */
usage_windows_by_feature_id?: Record<string, UsageWindow[]> | null;
/** Per-window deltas applied by this deduction (sibling stream of
* mutation_logs). */
usage_window_mutations?: UsageWindowMutation[];
remaining: number;
error?: string;
feature_id?: string;

14
server/src/internal/balances/utils/types/usageWindowMutation.ts Normal file
View File

@@ -0,0 +1,14 @@
/**
* One usage-window counter mutation from a deduction (sibling of
* MutationLogItem, kept as its own stream): which window row moved and by how
* much. The row is identified by its stored id plus the logical key
* (feature + window + entity scope); `usage_delta` is in the limit's native
* unit (tracked units for metered dims, credits for balance dims).
*/
export interface UsageWindowMutation {
usage_window_id: string | null;
feature_id: string;
internal_entity_id: string | null;
window_start_at: number;
usage_delta: number;
}

18
server/src/internal/balances/utils/types/usageWindowUpdate.ts Normal file
View File

@@ -0,0 +1,18 @@
import type { UsageWindow } from "@autumn/shared";
/**
* Post-deduction usage-window counter state for one capped feature, handed
* down from the Lua result through the deduction flow to syncItemV4 (sibling
* of DeductionUpdate / RolloverUpdate).
*
* Deliberately a SNAPSHOT, not a MutationLog-style delta: the deduction
* script is atomic and the Postgres sync full-replaces rows per (customer,
* feature), so the complete `usage_windows` array IS the update. An empty
* array is meaningful (all windows pruned/closed) and still full-replaces.
* Matches the `usage_window_updates` jsonb param of sync_balances_v2 1:1.
*/
export interface UsageWindowUpdate {
internal_customer_id: string;
feature_id: string;
usage_windows: UsageWindow[];
}

4
server/src/internal/billing/v2/compute/finalize/finalizeLineItems.ts
View File

@@ -31,6 +31,10 @@ export const finalizeLineItems = ({
autumnBillingPlan: AutumnBillingPlan;
customLineItems?: CustomLineItem[];
}): LineItem[] => {
if (billingContext.skipBillingChanges) {
return [];
}
if (
billingContext.requestedProrationBehavior === "none" &&
!billingContext.anchorResetRefund?.noPartialRefund

3
server/src/internal/billing/v2/providers/stripe/utils/checkoutSessions/updateOneOffTieredItems.ts
View File

@@ -1,4 +1,5 @@
import {
atmnToStripeAmount,
InternalError,
type LineItemContext,
type Organization,
@@ -76,7 +77,7 @@ export const updateOneOffTieredItems = ({
product_data: {
name: lineItem.description,
},
unit_amount: Math.round(lineItem.amount * 100),
unit_amount: atmnToStripeAmount({ amount: lineItem.amount, currency }),
currency,
},
quantity: 1,

4
server/src/internal/billing/v2/providers/stripe/utils/invoiceLines/lineItemsToCreateInvoiceItemsParams.ts
View File

@@ -35,11 +35,11 @@ const toStripeCreateInvoiceItemParams = ({
amount: shouldUsePriceData
? undefined
: atmnToStripeAmount({ amount: lineAmount }),
: atmnToStripeAmount({ amount: lineAmount, currency }),
price_data: shouldUsePriceData
? {
unit_amount: atmnToStripeAmount({ amount: lineAmount }),
unit_amount: atmnToStripeAmount({ amount: lineAmount, currency }),
currency,
product: stripeProductId,
}

4
server/src/internal/billing/v2/providers/stripe/utils/invoiceLines/lineItemsToInvoiceAddLinesParams.ts
View File

@@ -28,10 +28,10 @@ const toStripeAddLineParams = ({
description,
amount: shouldUsePriceData
? undefined
: atmnToStripeAmount({ amount: lineAmount }),
: atmnToStripeAmount({ amount: lineAmount, currency }),
price_data: shouldUsePriceData
? {
unit_amount: atmnToStripeAmount({ amount: lineAmount }),
unit_amount: atmnToStripeAmount({ amount: lineAmount, currency }),
currency,
product: stripeProductId,
}

5
server/src/internal/billing/v2/providers/stripe/utils/invoiceLines/lineItemsToSubscriptionAddInvoiceItemsParams.ts
View File

@@ -19,7 +19,10 @@ const toStripeSubscriptionAddInvoiceItem = ({
price_data: {
currency: context.currency,
product: stripeProductId,
unit_amount: atmnToStripeAmount({ amount: amountAfterDiscounts }),
unit_amount: atmnToStripeAmount({
amount: amountAfterDiscounts,
currency: context.currency,
}),
},
period: context.effectivePeriod
? {

20
server/src/internal/billing/v2/providers/stripe/utils/invoices/createInvoiceForBilling.ts
View File

@@ -1,7 +1,8 @@
import type {
BillingContext,
StripeDiscountWithCoupon,
StripeInvoiceAction,
import {
type BillingContext,
orgToCurrency,
type StripeDiscountWithCoupon,
type StripeInvoiceAction,
} from "@autumn/shared";
import {
type PayInvoiceResult,
@@ -92,12 +93,17 @@ export const createInvoiceForBilling = async ({
});
const wantsAutoTax = shouldEnableStripeAutomaticTax({ ctx, billingContext });
const stripeSubId = options.skipSubscriptionLink
? undefined
: billingContext.stripeSubscription?.id;
const draftInvoice = await createStripeInvoice({
stripeCli,
stripeCusId: billingContext.stripeCustomer?.id ?? "none",
stripeSubId: options.skipSubscriptionLink
? undefined
: billingContext.stripeSubscription?.id,
stripeSubId,
// Subscription-linked invoices inherit currency from the subscription;
// standalone invoices default to the account currency, not the org's.
currency: stripeSubId ? undefined : orgToCurrency({ org: ctx.org }),
collectionMethod,
daysUntilDue: invoiceMode?.daysUntilDue,
footer: invoiceMode?.footer,

18
server/src/internal/billing/v2/setup/setupBillingCycleAnchor.ts
View File

@@ -78,5 +78,21 @@ export const setupBillingCycleAnchor = ({
// Billing cycle anchor = trial ends at if exists
if (newIsTrialing) return trialContext?.trialEndsAt ?? "now";
return secondsToMs(stripeSubscription?.billing_cycle_anchor) ?? "now";
const stripeAnchorMs = secondsToMs(stripeSubscription?.billing_cycle_anchor);
// Stripe stores the anchor in SECONDS (rounded either way from the ms
// instant it was created). When it's the same instant the current product
// started, prefer the ms-precision starts_at so cycles recomputed across
// updates/upgrades don't drift sub-second (which would churn
// next_reset_at and spuriously move cycle-keyed state like usage windows).
const startsAtMs = customerProduct?.starts_at;
if (
stripeAnchorMs != null &&
startsAtMs != null &&
Math.abs(startsAtMs - stripeAnchorMs) < 1000
) {
return startsAtMs;
}
return stripeAnchorMs ?? "now";
};

2
server/src/internal/billing/v2/utils/lineItems/chargeRowToRefundLineItem.ts
View File

@@ -95,7 +95,7 @@ export const chargeRowToRefundLineItem = ({
context,
stripePriceId: chargeRow.stripe_price_id ?? undefined,
stripeProductId: chargeRow.stripe_product_id ?? undefined,
chargeImmediately: true,
chargeImmediately: chargeRow.invoice_id === null ? false : true,
prorated: true,
discounts:
(chargeRow.discounts as InvoiceLineItemDiscount[] | null)?.map((d) => ({

3
server/src/internal/billing/v2/utils/lineItems/getRefundLineItems.ts
View File

@@ -9,12 +9,14 @@ export const getRefundLineItems = ({
billingContext,
priceFilters,
billingCycleAnchorMsOverride,
includeCatalogFallback = true,
}: {
ctx: AutumnContext;
customerProduct: FullCusProduct;
billingContext: BillingContext;
priceFilters?: { excludeOneOffPrices?: boolean };
billingCycleAnchorMsOverride?: BillingContext["billingCycleAnchorMs"];
includeCatalogFallback?: boolean;
}): LineItem[] => {
const {
lineItems: matchedCredits,
@@ -27,6 +29,7 @@ export const getRefundLineItems = ({
});
if (allPricesResolved) return matchedCredits;
if (!includeCatalogFallback) return matchedCredits;
const catalogCredits = customerProductToLineItems({
ctx,

1
server/src/internal/billing/v2/utils/lineItems/getRefundLineItemsForPrice.ts
View File

@@ -19,6 +19,7 @@ export const getRefundLineItemsForPrice = ({
ctx,
customerProduct,
billingContext,
includeCatalogFallback: false,
});
const matchedRefundsForPrice = matchedRefundLineItems.filter(

4
server/src/internal/billing/v2/utils/lineItems/invoiceCreditFromStoredLineItems.ts
View File

@@ -58,7 +58,7 @@ export const invoiceCreditFromStoredLineItems = ({
row.customer_product_ids.length > 0 &&
row.effective_period_start != null &&
row.effective_period_end != null &&
row.effective_period_start < now &&
row.effective_period_start <= now &&
row.effective_period_end > now,
);
@@ -77,7 +77,7 @@ export const invoiceCreditFromStoredLineItems = ({
r.customer_product_ids.includes(customerProduct.id) &&
r.effective_period_end != null &&
r.effective_period_start != null &&
r.effective_period_start < now &&
r.effective_period_start <= now &&
r.effective_period_end > now,
);

34
server/src/internal/customers/actions/resetUsageWindows/applyUsageWindowRollsToSubject.ts Normal file
View File

@@ -0,0 +1,34 @@
import type { FullSubject, NormalizedFullSubject } from "@autumn/shared";
import type { UsageWindowRoll } from "./computeUsageWindowRolls.js";
/** Mirrors persisted rolls onto the in-flight subject (and its normalized
* twin), so this request's response already shows the rolled state. */
export const applyUsageWindowRollsToSubject = ({
fullSubject,
normalized,
rolls,
now,
}: {
fullSubject: FullSubject;
normalized?: NormalizedFullSubject;
rolls: UsageWindowRoll[];
now: number;
}): void => {
const rollsById = new Map(rolls.map((roll) => [roll.id, roll]));
const apply = (windows: FullSubject["usage_windows"]) => {
for (const usageWindow of windows ?? []) {
const roll = rollsById.get(usageWindow.id);
if (!roll) continue;
if (roll.zero_usage) usageWindow.usage = 0;
usageWindow.window_start_at = roll.window_start_at;
usageWindow.window_end_at = roll.window_end_at;
usageWindow.anchor_customer_entitlement_id =
roll.anchor_customer_entitlement_id;
usageWindow.updated_at = now;
}
};
apply(fullSubject.usage_windows);
if (normalized) apply(normalized.usage_windows);
};

87
server/src/internal/customers/actions/resetUsageWindows/computeUsageWindowRolls.ts Normal file
View File

@@ -0,0 +1,87 @@
import {
findUsageWindowLimitByWindow,
type UsageWindow,
type UsageWindowLimit,
} from "@autumn/shared";
export type UsageWindowRoll = {
id: string;
feature_id: string;
internal_entity_id: string | null;
/** True when the stored window closed: the count must zero. A roll never
* writes a count otherwise, so it can't clobber a concurrent deduction. */
zero_usage: boolean;
window_start_at: number;
window_end_at: number;
anchor_customer_entitlement_id: string | null;
};
/**
* Decides, per counter row, whether it needs rolling. A count is only valid
* within the exact window stamped on it; the anchor is provenance, so an
* anchor-only re-point keeps the count:
*
* expired | window moved | anchor moved | result
* --------+--------------+--------------+---------------------------------
* no | no | no | no roll (the common case)
* no | no | yes | re-point anchor, count kept
* no | yes | any | re-bound, count zeroed (plan change)
* yes | any | any | re-bound, count zeroed (period over)
* yes | (no limit) | -- | bounds kept, count zeroed (entity rows, v1)
*
* "Window moved" compares the row's bounds against its limit's CURRENT
* derivation (anchor ent's cycle). Entity-scoped rows have no resolvable
* limit in v1, so their bounds can't re-derive -- but an expired count must
* still zero.
*/
export const computeUsageWindowRolls = ({
usageWindows,
limits,
now,
}: {
usageWindows: UsageWindow[];
limits: UsageWindowLimit[];
now: number;
}): UsageWindowRoll[] => {
const rolls: UsageWindowRoll[] = [];
for (const usageWindow of usageWindows) {
const expired = Number(usageWindow.window_end_at) <= now;
const limit = findUsageWindowLimitByWindow({ limits, usageWindow });
const target = limit
? {
window_start_at: limit.window_start_at,
window_end_at: limit.window_end_at,
anchor_customer_entitlement_id: limit.anchor_customer_entitlement_id,
}
: {
window_start_at: Number(usageWindow.window_start_at),
window_end_at: Number(usageWindow.window_end_at),
anchor_customer_entitlement_id:
usageWindow.anchor_customer_entitlement_id ?? null,
};
const windowMoved =
Number(usageWindow.window_start_at) !== target.window_start_at ||
Number(usageWindow.window_end_at) !== target.window_end_at;
const anchorMoved =
(usageWindow.anchor_customer_entitlement_id ?? null) !==
target.anchor_customer_entitlement_id;
if (!expired && !windowMoved && !anchorMoved) continue;
rolls.push({
id: usageWindow.id,
feature_id: usageWindow.feature_id,
internal_entity_id: usageWindow.internal_entity_id ?? null,
// A count never survives its stamped window; an anchor-only
// re-point (e.g. an ent recreated with the same cycle) keeps it.
zero_usage: expired || windowMoved,
...target,
});
}
return rolls;
};

76
server/src/internal/customers/actions/resetUsageWindows/lazyResetSubjectUsageWindows.ts Normal file
View File

@@ -0,0 +1,76 @@
import {
type FullSubject,
fullSubjectToUsageWindowLimits,
type NormalizedFullSubject,
orgToInStatuses,
} from "@autumn/shared";
import * as Sentry from "@sentry/bun";
import { getDbHealth, PgHealth } from "@/db/pgHealthMonitor.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { usageWindowRepo } from "@/internal/customers/usageWindows/repos/index.js";
import { applyUsageWindowRollsToSubject } from "./applyUsageWindowRollsToSubject.js";
import { computeUsageWindowRolls } from "./computeUsageWindowRolls.js";
import { rollUsageWindowsCache } from "./rollUsageWindowsCache.js";
/**
* Lazily ROLLS the subject's usage-window counters on every subject read:
* zero counts whose stored window closed, and re-align bounds/anchor to the
* current derivation (this is where a plan change lands in the DB). The
* decision table lives in computeUsageWindowRolls.
*
* Best-effort, like lazyResetSubjectEntitlements: reads and the deduction
* script both derive a closed count as 0 and stamp fresh bounds on write, so
* a failed roll only delays persistence. Rolls are idempotent (same target
* state), so concurrent reads converge. Returns true if any rows rolled.
*/
export const lazyResetSubjectUsageWindows = async ({
ctx,
fullSubject,
normalized,
}: {
ctx: AutumnContext;
fullSubject: FullSubject;
normalized?: NormalizedFullSubject;
}): Promise<boolean> => {
if (getDbHealth() === PgHealth.Degraded) return false;
const now = Date.now();
const usageWindows = fullSubject.usage_windows ?? [];
if (usageWindows.length === 0) return false;
try {
const limits = fullSubjectToUsageWindowLimits({
fullSubject,
featureIds: [
...new Set(usageWindows.map((usageWindow) => usageWindow.feature_id)),
],
features: ctx.features,
now,
inStatuses: orgToInStatuses({ org: ctx.org }),
});
const rolls = computeUsageWindowRolls({ usageWindows, limits, now });
if (rolls.length === 0) return false;
ctx.logger.info(
`[lazyResetSubjectUsageWindows] customer: ${fullSubject.customerId}, rolling: ${rolls.length}`,
);
await usageWindowRepo.rollWindows({ db: ctx.db, rolls, now });
await rollUsageWindowsCache({
ctx,
customerId: fullSubject.customerId,
rolls,
now,
});
applyUsageWindowRollsToSubject({ fullSubject, normalized, rolls, now });
return true;
} catch (error) {
ctx.logger.error(
`[lazyResetSubjectUsageWindows] customer: ${fullSubject.customerId}, failed: ${error}`,
);
Sentry.captureException(error);
return false;
}
};

60
server/src/internal/customers/actions/resetUsageWindows/rollUsageWindowsCache.ts Normal file
View File

@@ -0,0 +1,60 @@
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
import { FULL_SUBJECT_CACHE_TTL_SECONDS } from "@/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.js";
import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
import type { UsageWindowRoll } from "./computeUsageWindowRolls.js";
/**
* Atomically patches rolled counters into each affected feature's
* '_usage_windows' field (one rollUsageWindows Lua call per feature,
* pipelined). Fire-and-forget -- reads and the deduction script both derive
* a closed window as 0, so a missed patch only delays the persisted roll.
*/
export const rollUsageWindowsCache = async ({
ctx,
customerId,
rolls,
now,
}: {
ctx: AutumnContext;
customerId: string;
rolls: UsageWindowRoll[];
now: number;
}): Promise<void> => {
if (rolls.length === 0) return;
try {
const { org, env, redisV2 } = ctx;
const rollsByFeatureId: Record<string, UsageWindowRoll[]> = {};
for (const roll of rolls) {
const featureRolls = rollsByFeatureId[roll.feature_id] ?? [];
featureRolls.push(roll);
rollsByFeatureId[roll.feature_id] = featureRolls;
}
const pipeline = redisV2.pipeline();
for (const [featureId, featureRolls] of Object.entries(rollsByFeatureId)) {
const balanceKey = buildSharedFullSubjectBalanceKey({
orgId: org.id,
env,
customerId,
featureId,
});
pipeline.rollUsageWindows(
balanceKey,
JSON.stringify({
now,
ttl_seconds: FULL_SUBJECT_CACHE_TTL_SECONDS,
rolls: featureRolls,
}),
);
}
await tryRedisWrite(() => pipeline.exec(), redisV2);
} catch (error) {
ctx.logger.error(
`[rollUsageWindowsCache] customer=${customerId}, failed: ${error}`,
);
}
};

2
server/src/internal/customers/actions/update/updateCustomer.ts
View File

@@ -139,6 +139,8 @@ export const updateCustomer = async ({
billingControlUpdates.auto_topups = billing_controls.auto_topups;
if (billing_controls.spend_limits !== undefined)
billingControlUpdates.spend_limits = billing_controls.spend_limits;
if (billing_controls.usage_limits !== undefined)
billingControlUpdates.usage_limits = billing_controls.usage_limits;
if (billing_controls.usage_alerts !== undefined)
billingControlUpdates.usage_alerts = billing_controls.usage_alerts;
if (billing_controls.overage_allowed !== undefined)

16
server/src/internal/customers/attach/attachFunctions/addProductFlow/handleOneOffFunction.ts
View File

@@ -2,6 +2,7 @@ import {
type AttachConfig,
type AttachFunctionResponse,
AttachFunctionResponseSchema,
atmnToStripeAmount,
isFixedPrice,
MetadataType,
priceToInvoiceAmount,
@@ -99,7 +100,10 @@ export const handleOneOffFunction = async ({
invoiceItemData = {
description,
price_data: {
unit_amount: new Decimal(amount).mul(100).round().toNumber(),
unit_amount: atmnToStripeAmount({
amount,
currency: orgToCurrency({ org }),
}),
currency: orgToCurrency({ org }),
product: price.config?.stripe_product_id || product?.processor?.id,
},
@@ -136,7 +140,7 @@ export const handleOneOffFunction = async ({
// Skip auto_tax in invoice mode: send_invoice has no
// address-collection UI so Stripe Tax rejects.
const wantsAutoTax =
const wantsAutoTax =
!!org.config.automatic_tax &&
!attachParams.invoiceOnly &&
customerHasUsableTaxLocationForStripeTax(attachParams.stripeCus);
@@ -145,12 +149,8 @@ const wantsAutoTax =
customer: customer.processor.id!,
auto_advance: false,
currency: orgToCurrency({ org }),
discounts: rewards
? rewards.map((r) => ({ coupon: r.id }))
: undefined,
collection_method: attachParams.invoiceOnly
? "send_invoice"
: undefined,
discounts: rewards ? rewards.map((r) => ({ coupon: r.id })) : undefined,
collection_method: attachParams.invoiceOnly ? "send_invoice" : undefined,
days_until_due: attachParams.invoiceOnly ? 30 : undefined,
...(shouldMemo ? { description: invoiceMemo } : {}),
...(wantsAutoTax ? { automatic_tax: { enabled: true } } : {}),

6
server/src/internal/customers/attach/attachFunctions/upgradeDiffIntFlow/createUsageInvoiceItems.ts
View File

@@ -1,4 +1,5 @@
import {
atmnToStripeAmount,
type BillingInterval,
BillingType,
cusProductsToCusPrices,
@@ -91,7 +92,10 @@ const getUsageInvoiceItems = async ({
description,
price_data: {
product: config.stripe_product_id!,
unit_amount: Math.round(amount * 100),
unit_amount: atmnToStripeAmount({
amount,
currency: org.default_currency || "usd",
}),
currency: org.default_currency || "usd",
},
period: {

6
server/src/internal/customers/attach/attachUtils/getContUseItems/createContUseInvoiceItems.ts
View File

@@ -1,4 +1,5 @@
import {
atmnToStripeAmount,
type BillingInterval,
BillingType,
cusProductToPrices,
@@ -138,7 +139,10 @@ export const createAndFilterContUseItems = async ({
const { start, end } = subToPeriodStartEnd({ sub });
await stripeCli.invoiceItems.create({
customer: customer.processor?.id ?? undefined,
amount: Math.round(item.amount * 100),
amount: atmnToStripeAmount({
amount: item.amount,
currency: org.default_currency || "usd",
}),
description: item.description,
currency: org.default_currency || "usd",
subscription: sub.id,

21
server/src/internal/customers/cache/fullSubject/actions/getCachedFullSubject.ts vendored
View File

@@ -7,10 +7,12 @@ import { isRedisMigrationCacheStale } from "@/external/redis/customerRedisRoutin
import { runRedisOp } from "@/external/redis/utils/runRedisOp.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { lazyResetSubjectEntitlements } from "@/internal/customers/actions/resetCustomerEntitlementsV2/lazyResetSubjectEntitlements.js";
import { lazyResetSubjectUsageWindows } from "@/internal/customers/actions/resetUsageWindows/lazyResetSubjectUsageWindows.js";
import { checkPendingMigrationsForCustomer } from "@/internal/migrations/v2/lazy/checkPendingMigrationsForCustomer.js";
import { getFullSubjectRolloutSnapshot } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
import { isSnapshotCacheStale } from "@/internal/misc/rollouts/rolloutUtils.js";
import { applyLiveAggregatedBalances } from "../balances/applyLiveAggregatedBalances.js";
import { applyLiveUsageWindows } from "../balances/applyLiveUsageWindows.js";
import { getCachedFeatureBalancesBatch } from "../balances/getCachedFeatureBalances.js";
import { buildFullSubjectKey } from "../builders/buildFullSubjectKey.js";
import { buildFullSubjectViewEpochKey } from "../builders/buildFullSubjectViewEpochKey.js";
@@ -195,12 +197,19 @@ export const getCachedFullSubject = async ({
}
const isCustomerSubject = !entityId;
// Capped features may have no entitlements, so they aren't guaranteed to be
// in meteredFeatures; union them in so their `_usage_windows` field is read.
const usageWindowFeatureIds = new Set(cached.usageWindowFeatureIds ?? []);
const batchFeatureIds = [
...new Set([...cached.meteredFeatures, ...usageWindowFeatureIds]),
];
const balancesOutcome = await getCachedFeatureBalancesBatch({
ctx,
customerId,
featureIds: cached.meteredFeatures,
featureIds: batchFeatureIds,
customerEntitlementIdsByFeatureId: cached.customerEntitlementIdsByFeatureId,
includeAggregated: isCustomerSubject,
usageWindowFeatureIds,
});
if (balancesOutcome.kind === "missing") {
@@ -220,9 +229,9 @@ export const getCachedFullSubject = async ({
}
const balances = balancesOutcome.value;
if (balances.length !== cached.meteredFeatures.length) {
if (balances.length !== batchFeatureIds.length) {
logger.warn(
`[getCachedFullSubject] Incomplete cache for ${customerId}${entityId ? `:${entityId}` : ""}: expected ${cached.meteredFeatures.length} balance keys, got ${balances.length}. Rebuilding from DB, source: ${source}`,
`[getCachedFullSubject] Incomplete cache for ${customerId}${entityId ? `:${entityId}` : ""}: expected ${batchFeatureIds.length} balance keys, got ${balances.length}. Rebuilding from DB, source: ${source}`,
);
await invalidateCachedFullSubjectExact({
ctx,
@@ -249,8 +258,14 @@ export const getCachedFullSubject = async ({
});
}
applyLiveUsageWindows({
normalized,
featureBalances: balances,
});
const fullSubject = normalizedToFullSubject({ normalized });
await lazyResetSubjectEntitlements({ ctx, fullSubject });
await lazyResetSubjectUsageWindows({ ctx, fullSubject, normalized });
await checkPendingMigrationsForCustomer({
ctx,
fullCustomer: fullSubjectToFullCustomer({ fullSubject }),

1
server/src/internal/customers/cache/fullSubject/actions/getOrSetCachedFullSubject.ts vendored
View File

@@ -27,7 +27,6 @@ export const getOrSetCachedFullSubject = async ({
let fetchedSubjectViewEpoch = 0;
if (useRedis) {
// The pipeline inside getCachedFullSubject already fetches + refreshes
// the epoch, so we reuse it on miss instead of a second round trip.

31
server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateSharedBalanceFields.ts vendored
View File

@@ -3,7 +3,10 @@ import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
import { buildSharedFullSubjectBalanceKey } from "../../builders/buildSharedFullSubjectBalanceKey.js";
import { AGGREGATED_BALANCE_FIELD } from "../../config/fullSubjectCacheConfig.js";
import {
AGGREGATED_BALANCE_FIELD,
USAGE_WINDOWS_FIELD,
} from "../../config/fullSubjectCacheConfig.js";
import type { CachedFullSubject } from "../../fullSubjectCacheModel.js";
/**
@@ -61,19 +64,37 @@ async function deleteFieldsFromManifest({
const { customerEntitlementIdsByFeatureId } = manifest;
if (!customerEntitlementIdsByFeatureId) return;
// Capped features may have no entitlements, so their hashes only appear in
// usageWindowFeatureIds; union both so `_usage_windows` is cleared too.
// Safe to delete counters: capped tracks write through to PG synchronously,
// and the rebuild re-seeds the field from PG.
// Raw blob, no sanitize walker: cjson re-encodes empty arrays as {}, so
// array fields must be Array.isArray-guarded before spreading.
const usageWindowFeatureIds = Array.isArray(manifest.usageWindowFeatureIds)
? manifest.usageWindowFeatureIds
: [];
const featureIds = new Set([
...Object.keys(customerEntitlementIdsByFeatureId),
...usageWindowFeatureIds,
]);
const pipeline = redisV2.pipeline();
let fieldCount = 0;
for (const [featureId, cusEntIds] of Object.entries(
customerEntitlementIdsByFeatureId,
)) {
for (const featureId of featureIds) {
const rawCusEntIds = customerEntitlementIdsByFeatureId[featureId];
const cusEntIds = Array.isArray(rawCusEntIds) ? rawCusEntIds : [];
const balanceKey = buildSharedFullSubjectBalanceKey({
orgId: org.id,
env,
customerId,
featureId,
});
const fieldsToDelete = [...cusEntIds, AGGREGATED_BALANCE_FIELD];
const fieldsToDelete = [
...cusEntIds,
AGGREGATED_BALANCE_FIELD,
USAGE_WINDOWS_FIELD,
];
pipeline.hdel(balanceKey, ...fieldsToDelete);
fieldCount += fieldsToDelete.length;
}

26
server/src/internal/customers/cache/fullSubject/actions/partial/getCachedPartialFullSubject.ts vendored
View File

@@ -4,9 +4,11 @@ import { isRedisMigrationCacheStale } from "@/external/redis/customerRedisRoutin
import { runRedisOp } from "@/external/redis/utils/runRedisOp.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { lazyResetSubjectEntitlements } from "@/internal/customers/actions/resetCustomerEntitlementsV2/lazyResetSubjectEntitlements.js";
import { lazyResetSubjectUsageWindows } from "@/internal/customers/actions/resetUsageWindows/lazyResetSubjectUsageWindows.js";
import { getFullSubjectRolloutSnapshot } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
import { isSnapshotCacheStale } from "@/internal/misc/rollouts/rolloutUtils.js";
import { applyLiveAggregatedBalances } from "../../balances/applyLiveAggregatedBalances.js";
import { applyLiveUsageWindows } from "../../balances/applyLiveUsageWindows.js";
import { getCachedFeatureBalancesBatch } from "../../balances/getCachedFeatureBalances.js";
import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
import { buildFullSubjectViewEpochKey } from "../../builders/buildFullSubjectViewEpochKey.js";
@@ -212,9 +214,22 @@ export const getCachedPartialFullSubject = async ({
};
}
const meteredFeatureIdsToFetch = featureIds.filter((featureId) =>
cached.meteredFeatures.includes(featureId),
// Capped features carry the '_usage_windows' counter field and may have no
// entitlements at all, so they must be part of the batch even when absent
// from meteredFeatures.
const usageWindowFeatureIds = new Set(
(cached.usageWindowFeatureIds ?? []).filter((featureId) =>
featureIds.includes(featureId),
),
);
const meteredFeatureIdsToFetch = [
...new Set([
...featureIds.filter((featureId) =>
cached.meteredFeatures.includes(featureId),
),
...usageWindowFeatureIds,
]),
];
const isCustomerSubject = !entityId;
const featureBalancesOutcome = await getCachedFeatureBalancesBatch({
@@ -223,6 +238,7 @@ export const getCachedPartialFullSubject = async ({
featureIds: meteredFeatureIdsToFetch,
customerEntitlementIdsByFeatureId: cached.customerEntitlementIdsByFeatureId,
includeAggregated: isCustomerSubject,
usageWindowFeatureIds,
});
const invalidateIncomplete = () =>
@@ -287,8 +303,14 @@ export const getCachedPartialFullSubject = async ({
});
}
applyLiveUsageWindows({
normalized,
featureBalances,
});
const fullSubject = normalizedToFullSubject({ normalized });
await lazyResetSubjectEntitlements({ ctx, fullSubject, normalized });
await lazyResetSubjectUsageWindows({ ctx, fullSubject, normalized });
return fullSubject;
},
invalidate: () =>

20
server/src/internal/customers/cache/fullSubject/actions/rehydrateWithLiveBalances.ts vendored
View File

@@ -2,6 +2,7 @@ import type { NormalizedFullSubject } from "@autumn/shared";
import { type FullSubject, normalizedToFullSubject } from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { applyLiveAggregatedBalances } from "../balances/applyLiveAggregatedBalances.js";
import { applyLiveUsageWindows } from "../balances/applyLiveUsageWindows.js";
import { getCachedFeatureBalancesBatch } from "../balances/getCachedFeatureBalances.js";
/**
@@ -31,7 +32,18 @@ export const rehydrateWithLiveBalances = async ({
list.push(ce.id);
customerEntitlementIdsByFeatureId[ce.feature_id] = list;
}
const featureIds = Object.keys(customerEntitlementIdsByFeatureId);
const usageWindowFeatureIds = new Set(
[
...(normalized.customer.usage_limits ?? []),
...(normalized.entity?.usage_limits ?? []),
].map((usageLimit) => usageLimit.feature_id),
);
const featureIds = [
...new Set([
...Object.keys(customerEntitlementIdsByFeatureId),
...usageWindowFeatureIds,
]),
];
const isCustomerSubject = !entityId;
const outcome = await getCachedFeatureBalancesBatch({
@@ -40,6 +52,7 @@ export const rehydrateWithLiveBalances = async ({
featureIds,
customerEntitlementIdsByFeatureId,
includeAggregated: isCustomerSubject,
usageWindowFeatureIds,
});
if (outcome.kind !== "ok") return undefined;
@@ -53,5 +66,10 @@ export const rehydrateWithLiveBalances = async ({
});
}
applyLiveUsageWindows({
normalized,
featureBalances: outcome.value,
});
return normalizedToFullSubject({ normalized });
};

2
server/src/internal/customers/cache/fullSubject/actions/setCachedFullSubject/setCachedFullSubject.ts vendored
View File

@@ -49,6 +49,8 @@ export const setCachedFullSubject = async ({
customerEntitlements: normalized.customer_entitlements,
aggregatedCustomerEntitlements:
normalized.entity_aggregations?.aggregated_customer_entitlements ?? [],
usageWindows: normalized.usage_windows ?? [],
usageWindowFeatureIds: cached.usageWindowFeatureIds,
});
const keys: string[] = [subjectKey, epochKey];

30
server/src/internal/customers/cache/fullSubject/actions/setCachedFullSubject/setSharedFullSubjectBalances.ts vendored
View File

@@ -1,10 +1,14 @@
import type {
AggregatedFeatureBalance,
NormalizedFullSubject,
UsageWindow,
} from "@autumn/shared";
import { featureBalancesToHashFields } from "../../balances/featureBalancesToHashFields.js";
import { buildSharedFullSubjectBalanceKey } from "../../builders/buildSharedFullSubjectBalanceKey.js";
import { AGGREGATED_BALANCE_FIELD } from "../../config/fullSubjectCacheConfig.js";
import {
AGGREGATED_BALANCE_FIELD,
USAGE_WINDOWS_FIELD,
} from "../../config/fullSubjectCacheConfig.js";
export type SharedBalanceWrite = {
balanceKey: string;
@@ -17,12 +21,16 @@ export const buildSharedBalanceWrites = ({
customerId,
customerEntitlements,
aggregatedCustomerEntitlements,
usageWindows = [],
usageWindowFeatureIds = [],
}: {
orgId: string;
env: string;
customerId: string;
customerEntitlements: NormalizedFullSubject["customer_entitlements"];
aggregatedCustomerEntitlements: AggregatedFeatureBalance[];
usageWindows?: UsageWindow[];
usageWindowFeatureIds?: string[];
}): SharedBalanceWrite[] => {
const balancesByFeatureId = new Map<string, typeof customerEntitlements>();
@@ -38,9 +46,24 @@ export const buildSharedBalanceWrites = ({
aggregatedByFeatureId.set(aggregated.feature_id, aggregated);
}
// Capped features get a `_usage_windows` field even with no rows and no
// entitlements: a present-but-empty field means "fresh counter", a missing
// field means "stale cache" and the deduction script fails closed on it.
const usageWindowsByFeatureId = new Map<string, UsageWindow[]>();
for (const featureId of usageWindowFeatureIds) {
usageWindowsByFeatureId.set(featureId, []);
}
for (const usageWindow of usageWindows) {
const existingWindows = usageWindowsByFeatureId.get(usageWindow.feature_id);
// Rows for features whose cap is no longer armed are not re-cached.
if (!existingWindows) continue;
existingWindows.push(usageWindow);
}
const allFeatureIds = new Set([
...balancesByFeatureId.keys(),
...aggregatedByFeatureId.keys(),
...usageWindowsByFeatureId.keys(),
]);
return Array.from(allFeatureIds).map((featureId) => {
@@ -52,6 +75,11 @@ export const buildSharedBalanceWrites = ({
fields[AGGREGATED_BALANCE_FIELD] = JSON.stringify(aggregated);
}
const featureUsageWindows = usageWindowsByFeatureId.get(featureId);
if (featureUsageWindows) {
fields[USAGE_WINDOWS_FIELD] = JSON.stringify(featureUsageWindows);
}
return {
balanceKey: buildSharedFullSubjectBalanceKey({
orgId,

20
server/src/internal/customers/cache/fullSubject/balances/applyLiveUsageWindows.ts vendored Normal file
View File

@@ -0,0 +1,20 @@
import type { NormalizedFullSubject } from "@autumn/shared";
import type { FeatureBalanceResult } from "./getCachedFeatureBalances.js";
/**
* Fill `normalized.usage_windows` from the live `_usage_windows` hash fields
* returned by the batch balance read. The cached subject view never carries
* counter rows (they'd be instantly stale), so this is the only hydration
* source on the cache-hit path.
*/
export const applyLiveUsageWindows = ({
normalized,
featureBalances,
}: {
normalized: NormalizedFullSubject;
featureBalances: FeatureBalanceResult[];
}): void => {
normalized.usage_windows = featureBalances.flatMap(
(featureBalance) => featureBalance.usageWindows ?? [],
);
};

54
server/src/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.ts vendored
View File

@@ -1,8 +1,15 @@
import type { AggregatedFeatureBalance, SubjectBalance } from "@autumn/shared";
import type {
AggregatedFeatureBalance,
SubjectBalance,
UsageWindow,
} from "@autumn/shared";
import { runRedisOp } from "@/external/redis/utils/runRedisOp.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { buildSharedFullSubjectBalanceKey } from "../builders/buildSharedFullSubjectBalanceKey.js";
import { AGGREGATED_BALANCE_FIELD } from "../config/fullSubjectCacheConfig.js";
import {
AGGREGATED_BALANCE_FIELD,
USAGE_WINDOWS_FIELD,
} from "../config/fullSubjectCacheConfig.js";
import { roundSubjectBalance } from "../roundCacheBalance.js";
import {
sanitizeCachedAggregatedFeatureBalance,
@@ -13,6 +20,24 @@ export type FeatureBalanceResult = {
featureId: string;
balances: SubjectBalance[];
aggregated?: AggregatedFeatureBalance;
/** Customer-scoped windowed-cap counters for this feature; only present for
* features in the requested usageWindowFeatureIds set. */
usageWindows?: UsageWindow[];
};
// Fail open: a missing/unparseable `_usage_windows` field reads as an empty
// counter set (the window restarts). cjson also encodes an empty Lua table as
// `{}`, so a non-array blob is an empty set, not corruption.
const parseUsageWindowsField = (
usageWindowsJson: string | null,
): UsageWindow[] => {
if (!usageWindowsJson) return [];
try {
const parsed = JSON.parse(usageWindowsJson);
return Array.isArray(parsed) ? (parsed as UsageWindow[]) : [];
} catch {
return [];
}
};
export type FeatureBalanceOutcome =
@@ -118,12 +143,16 @@ export const getCachedFeatureBalancesBatch = async ({
featureIds,
customerEntitlementIdsByFeatureId,
includeAggregated = false,
usageWindowFeatureIds,
}: {
ctx: AutumnContext;
customerId: string;
featureIds: string[];
customerEntitlementIdsByFeatureId: Record<string, string[]>;
includeAggregated?: boolean;
/** Features with an armed windowed cap: their `_usage_windows` field is
* read too. A missing field fails open (reads as an empty counter set). */
usageWindowFeatureIds?: Set<string>;
}): Promise<FeatureBalancesBatchOutcome> => {
if (featureIds.length === 0) return { kind: "ok", value: [] };
@@ -132,9 +161,11 @@ export const getCachedFeatureBalancesBatch = async ({
for (const featureId of featureIds) {
const customerEntitlementIds =
customerEntitlementIdsByFeatureId[featureId] ?? [];
const fields = includeAggregated
? [...customerEntitlementIds, AGGREGATED_BALANCE_FIELD]
: customerEntitlementIds;
const fields = [...customerEntitlementIds];
if (includeAggregated) fields.push(AGGREGATED_BALANCE_FIELD);
if (usageWindowFeatureIds?.has(featureId)) {
fields.push(USAGE_WINDOWS_FIELD);
}
pipeline.hmget(
buildSharedFullSubjectBalanceKey({
orgId: org.id,
@@ -167,7 +198,12 @@ export const getCachedFeatureBalancesBatch = async ({
};
let aggregated: AggregatedFeatureBalance | undefined;
let ceValues: (string | null)[];
let usageWindows: UsageWindow[] | undefined;
// Pop reserved fields in reverse push order: [_aggregated?, _usage_windows?].
if (usageWindowFeatureIds?.has(featureIds[i])) {
usageWindows = parseUsageWindowsField(allValues.pop() ?? null);
}
if (includeAggregated) {
const aggregatedJson = allValues.pop() ?? null;
@@ -181,11 +217,10 @@ export const getCachedFeatureBalancesBatch = async ({
// Malformed _aggregated is non-fatal; fall back to subject string value
}
}
ceValues = allValues;
} else {
ceValues = allValues;
}
const ceValues = allValues;
if (ceValues.length !== customerEntitlementIds.length)
return {
kind: "missing",
@@ -221,6 +256,7 @@ export const getCachedFeatureBalancesBatch = async ({
featureId: featureIds[i],
balances,
aggregated,
usageWindows,
});
}

20
server/src/internal/customers/cache/fullSubject/builders/buildDeductFromSubjectBalancesKeys.ts vendored
View File

@@ -21,6 +21,7 @@ export const buildDeductFromSubjectBalancesKeys = ({
idempotencyKey,
customerEntitlementDeductions,
fallbackFeatureId,
usageWindowFeatureIds = [],
}: {
orgId: string;
env: AppEnv;
@@ -30,17 +31,26 @@ export const buildDeductFromSubjectBalancesKeys = ({
idempotencyKey?: string | null;
customerEntitlementDeductions: { feature_id?: string }[];
fallbackFeatureId: string;
// Capped features: their balance hashes carry the `_usage_windows` counter
// field, and a capped feature may have no entitlements (so no deduction
// entry references its hash). Declare those keys in KEYS[] too.
usageWindowFeatureIds?: string[];
}) => {
const balanceKeysByFeatureId: Record<string, string> = {};
for (const deductionEntry of customerEntitlementDeductions) {
const targetFeatureId = deductionEntry.feature_id ?? fallbackFeatureId;
if (balanceKeysByFeatureId[targetFeatureId]) continue;
balanceKeysByFeatureId[targetFeatureId] = buildSharedFullSubjectBalanceKey({
const addFeatureKey = (featureId: string) => {
if (balanceKeysByFeatureId[featureId]) return;
balanceKeysByFeatureId[featureId] = buildSharedFullSubjectBalanceKey({
orgId,
env,
customerId,
featureId: targetFeatureId,
featureId,
});
};
for (const deductionEntry of customerEntitlementDeductions) {
addFeatureKey(deductionEntry.feature_id ?? fallbackFeatureId);
}
for (const usageWindowFeatureId of usageWindowFeatureIds) {
addFeatureKey(usageWindowFeatureId);
}
const balanceFeatureIds = Object.keys(balanceKeysByFeatureId);

6
server/src/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.ts vendored
View File

@@ -3,3 +3,9 @@ import { seconds } from "@autumn/shared";
export const FULL_SUBJECT_CACHE_TTL_SECONDS = seconds.days(3);
export const FULL_SUBJECT_EPOCH_TTL_SECONDS = seconds.days(5);
export const AGGREGATED_BALANCE_FIELD = "_aggregated";
// Customer-scoped usage-window counters for the capped feature, stored as a
// reserved field in that feature's balance hash (JSON array of rows). The
// rebuild writes it (even []) for armed caps; readers fail OPEN on a missing
// field (the window restarts), so it is a warm-read optimization, not a
// correctness contract.
export const USAGE_WINDOWS_FIELD = "_usage_windows";

27
server/src/internal/customers/cache/fullSubject/fullSubjectCacheModel.ts vendored
View File

@@ -16,14 +16,23 @@ import {
} from "@autumn/shared";
import { z } from "zod/v4";
// `usage_windows` is omitted alongside balances: counters live in the
// per-feature balance hashes (`_usage_windows` field) and would be instantly
// stale if serialized into the subject view.
export type CachedFullSubject = Omit<
NormalizedFullSubject,
"customer_entitlements"
"customer_entitlements" | "usage_windows"
> & {
_schemaVersion: number;
_cachedAt: number;
meteredFeatures: string[];
customerEntitlementIdsByFeatureId: Record<string, string[]>;
/** Features with an armed windowed cap (customer + entity usage_limits);
* may include features with no entitlements, so it cannot be derived from
* customerEntitlementIdsByFeatureId. Drives `_usage_windows` reads,
* writes, and invalidation. Optional: cache entries written before usage
* windows existed don't carry it (treat as []). */
usageWindowFeatureIds?: string[];
subjectViewEpoch: number;
};
@@ -72,6 +81,9 @@ export const CachedFullSubjectSchema = z.object({
_cachedAt: z.number(),
meteredFeatures: z.array(z.string()),
customerEntitlementIdsByFeatureId: z.record(z.string(), z.array(z.string())),
// Optional (not defaulted): pre-usage-windows cache entries don't carry it,
// and the hole-filling walker must not invent it.
usageWindowFeatureIds: z.array(z.string()).optional(),
subjectViewEpoch: z.number(),
});
@@ -105,6 +117,15 @@ export const normalizedToCachedFullSubject = ({
const meteredFeatures = [...meteredFeatureSet];
const usageWindowFeatureIds = [
...new Set(
[
...(normalized.customer.usage_limits ?? []),
...(normalized.entity?.usage_limits ?? []),
].map((usageLimit) => usageLimit.feature_id),
),
];
return {
subjectType: normalized.subjectType,
customerId: normalized.customerId,
@@ -128,6 +149,7 @@ export const normalizedToCachedFullSubject = ({
_cachedAt: Date.now(),
meteredFeatures,
customerEntitlementIdsByFeatureId,
usageWindowFeatureIds,
subjectViewEpoch,
};
};
@@ -159,5 +181,8 @@ export const cachedFullSubjectToNormalized = ({
invoices: cached.invoices,
entity_aggregations: cached.entity_aggregations,
migration_item_runs: cached.migration_item_runs ?? [],
// Live data: filled from the balance hashes' `_usage_windows` fields by
// the caller, never from the cached subject view.
usage_windows: [],
};
};

18
server/src/internal/customers/cache/fullSubject/roundCacheBalance.ts vendored
View File

@@ -5,9 +5,7 @@ import { Decimal } from "decimal.js";
* Round a number to avoid floating-point precision issues from Lua 5.1 double arithmetic.
* Uses Decimal.js toDecimalPlaces(10) — enough precision while eliminating float drift.
*/
export const roundCacheBalance = (
value: number | null | undefined,
): number => {
export const roundCacheBalance = (value: number | null | undefined): number => {
if (value === null || value === undefined) return 0;
return new Decimal(value).toDecimalPlaces(10).toNumber();
};
@@ -23,11 +21,19 @@ export const roundSubjectBalance = ({
}): SubjectBalance => {
subjectBalance.balance = roundCacheBalance(subjectBalance.balance);
if (subjectBalance.adjustment !== null && subjectBalance.adjustment !== undefined)
if (
subjectBalance.adjustment !== null &&
subjectBalance.adjustment !== undefined
)
subjectBalance.adjustment = roundCacheBalance(subjectBalance.adjustment);
if (subjectBalance.additional_balance !== null && subjectBalance.additional_balance !== undefined)
subjectBalance.additional_balance = roundCacheBalance(subjectBalance.additional_balance);
if (
subjectBalance.additional_balance !== null &&
subjectBalance.additional_balance !== undefined
)
subjectBalance.additional_balance = roundCacheBalance(
subjectBalance.additional_balance,
);
if (subjectBalance.entities && typeof subjectBalance.entities === "object") {
for (const entityId of Object.keys(subjectBalance.entities)) {

9
server/src/internal/customers/cusUtils/apiCusUtils/getApiCustomerBase.ts
View File

@@ -5,6 +5,9 @@ import {
CustomerExpand,
type CustomerLegacyData,
type FullCustomer,
fullCustomerToFullSubject,
fullSubjectToApiUsageLimits,
orgToInStatuses,
scopeExpandForCtx,
} from "@autumn/shared";
import { z } from "zod/v4";
@@ -45,6 +48,11 @@ export const getApiCustomerBase = async ({
ctx: subscriptionsScopedCtx,
fullCus,
});
const usageLimits = fullSubjectToApiUsageLimits({
fullSubject: fullCustomerToFullSubject({ fullCustomer: fullCus }),
features: ctx.features,
inStatuses: orgToInStatuses({ org: ctx.org }),
});
const apiCustomer = ApiCustomerV5Schema.extend({
autumn_id: z.string().optional(),
@@ -69,6 +77,7 @@ export const getApiCustomerBase = async ({
billing_controls: {
auto_topups: fullCus.auto_topups ?? undefined,
spend_limits: fullCus.spend_limits ?? undefined,
usage_limits: usageLimits,
usage_alerts: fullCus.usage_alerts ?? undefined,
overage_allowed: fullCus.overage_allowed ?? undefined,
},

Some files were not shown because too many files have changed in this diff Show More