Skip to content

Commit 1ea0111

Browse files
tech-adriantech-adrian
authored
Feat/anchor indexer batching sessions (#233)
* feat: add database migrations for batch processing, session management, and stellar event indexing - Create payment_batches, batch_items, and batch_status_history tables - Create user_sessions, session_activity_log, and session_security_events tables - Create stellar_events and indexer_state tables for real-time event monitoring - Add comprehensive indexes and triggers for automatic timestamp updates * feat: implement batch processing service - Create BatchService with batch creation, item management, and processing logic - Support configurable batch sizes and retry logic - Track batch status and generate processing reports - Implement partial failure handling and error tracking * feat: implement real-time stellar event monitoring - Connect to Stellar Horizon streaming API for transaction monitoring - Fetch and index payment and transfer events in real-time - Store events in database with proper status tracking - Implement event processing and payment reconciliation - Add reconnection logic with exponential backoff for reliability * feat: add batch processing HTTP endpoints and service integration - Create batch HTTP handlers for CRUD operations and processing - Add batch routes to API v1 and v2 - Integrate BatchService and SessionService into ServiceContainer - Add sha2 dependency for session token generation - Support batch creation, item management, and status reporting
1 parent d79381d commit 1ea0111

11 files changed

Lines changed: 1815 additions & 11 deletions

‎backend/Cargo.toml‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,12 @@ tokio-util = "0.7"
1111
# Web framework
1212
axum = { version = "0.7", features = ["json", "multipart", "macros"] }
1313
tower = { version = "0.4", features = ["util"] }
14-
tower-http = { version = "0.5", features = ["cors", "trace", "request-id", "util"] }
14+
tower-http = { version = "0.5", features = [
15+
"cors",
16+
"trace",
17+
"request-id",
18+
"util",
19+
] }
1520
governor = "0.6"
1621

1722
# Serialization
@@ -25,15 +30,20 @@ sqlx = { version = "0.7", features = [
2530
"chrono",
2631
"uuid",
2732
"json",
28-
"migrate"
33+
"migrate",
2934
] }
3035
deadpool-postgres = "0.10"
31-
tokio-postgres = { version = "0.7", features = ["with-chrono-0_4", "with-uuid-1", "with-serde_json-1"] }
36+
tokio-postgres = { version = "0.7", features = [
37+
"with-chrono-0_4",
38+
"with-uuid-1",
39+
"with-serde_json-1",
40+
] }
3241

3342
# Authentication & Security
3443
jsonwebtoken = "9.0"
3544
bcrypt = "0.15"
3645
ring = "0.17"
46+
sha2 = "0.10"
3747

3848
# Stellar SDK - TODO: Use correct Stellar Rust SDK
3949
# Check: https://crates.io/crates/stellar-rust-sdk or similar

‎backend/migrations/20260527000000_batch_processing.sql‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
-- Payment Batch Processing Service
2+
-- Creates tables for batch processing, tracking, and status management.
3+
4+
-- -------------------------------------------------------------------------
5+
-- payment_batches
6+
-- -------------------------------------------------------------------------
7+
CREATE TABLE IF NOT EXISTS payment_batches (
8+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
9+
batch_key VARCHAR(255) UNIQUE NOT NULL,
10+
merchant_id VARCHAR(255) NOT NULL,
11+
status VARCHAR(50) NOT NULL DEFAULT 'pending',
12+
total_amount BIGINT NOT NULL,
13+
total_count INTEGER NOT NULL,
14+
processed_count INTEGER NOT NULL DEFAULT 0,
15+
failed_count INTEGER NOT NULL DEFAULT 0,
16+
asset VARCHAR(56) NOT NULL,
17+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
18+
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
19+
completed_at TIMESTAMP WITH TIME ZONE,
20+
21+
CONSTRAINT fk_batch_merchant
22+
FOREIGN KEY (merchant_id) REFERENCES merchants(merchant_id) ON DELETE RESTRICT,
23+
24+
CONSTRAINT chk_batch_status CHECK (
25+
status IN ('pending', 'processing', 'completed', 'failed', 'partial_failure')
26+
),
27+
28+
CONSTRAINT chk_batch_counts CHECK (
29+
processed_count + failed_count <= total_count
30+
)
31+
);
32+
33+
-- -------------------------------------------------------------------------
34+
-- batch_items
35+
-- -------------------------------------------------------------------------
36+
CREATE TABLE IF NOT EXISTS batch_items (
37+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
38+
batch_id UUID NOT NULL,
39+
payment_id UUID NOT NULL,
40+
status VARCHAR(50) NOT NULL DEFAULT 'pending',
41+
error_message TEXT,
42+
retry_count INTEGER NOT NULL DEFAULT 0,
43+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
44+
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
45+
46+
CONSTRAINT fk_batch_item_batch
47+
FOREIGN KEY (batch_id) REFERENCES payment_batches(id) ON DELETE CASCADE,
48+
49+
CONSTRAINT fk_batch_item_payment
50+
FOREIGN KEY (payment_id) REFERENCES payments(id) ON DELETE RESTRICT,
51+
52+
CONSTRAINT chk_batch_item_status CHECK (
53+
status IN ('pending', 'processing', 'completed', 'failed')
54+
),
55+
56+
CONSTRAINT chk_retry_count CHECK (retry_count >= 0)
57+
);
58+
59+
-- -------------------------------------------------------------------------
60+
-- batch_status_history
61+
-- -------------------------------------------------------------------------
62+
CREATE TABLE IF NOT EXISTS batch_status_history (
63+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
64+
batch_id UUID NOT NULL,
65+
old_status VARCHAR(50),
66+
new_status VARCHAR(50) NOT NULL,
67+
reason TEXT,
68+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
69+
70+
CONSTRAINT fk_status_history_batch
71+
FOREIGN KEY (batch_id) REFERENCES payment_batches(id) ON DELETE CASCADE
72+
);
73+
74+
-- -------------------------------------------------------------------------
75+
-- Indexes
76+
-- -------------------------------------------------------------------------
77+
CREATE INDEX IF NOT EXISTS idx_batches_merchant_id
78+
ON payment_batches(merchant_id);
79+
80+
CREATE INDEX IF NOT EXISTS idx_batches_status
81+
ON payment_batches(status);
82+
83+
CREATE INDEX IF NOT EXISTS idx_batches_created_at
84+
ON payment_batches(created_at DESC);
85+
86+
CREATE INDEX IF NOT EXISTS idx_batch_items_batch_id
87+
ON batch_items(batch_id);
88+
89+
CREATE INDEX IF NOT EXISTS idx_batch_items_payment_id
90+
ON batch_items(payment_id);
91+
92+
CREATE INDEX IF NOT EXISTS idx_batch_items_status
93+
ON batch_items(status);
94+
95+
CREATE INDEX IF NOT EXISTS idx_status_history_batch_id
96+
ON batch_status_history(batch_id);
97+
98+
-- -------------------------------------------------------------------------
99+
-- Trigger: auto-update updated_at on payment_batches
100+
-- -------------------------------------------------------------------------
101+
CREATE OR REPLACE FUNCTION update_batch_updated_at()
102+
RETURNS TRIGGER AS $
103+
BEGIN
104+
NEW.updated_at = NOW();
105+
RETURN NEW;
106+
END;
107+
$ LANGUAGE plpgsql;
108+
109+
DROP TRIGGER IF EXISTS trg_batch_updated_at ON payment_batches;
110+
CREATE TRIGGER trg_batch_updated_at
111+
BEFORE UPDATE ON payment_batches
112+
FOR EACH ROW
113+
EXECUTE FUNCTION update_batch_updated_at();
114+
115+
-- -------------------------------------------------------------------------
116+
-- Trigger: auto-update updated_at on batch_items
117+
-- -------------------------------------------------------------------------
118+
CREATE OR REPLACE FUNCTION update_batch_item_updated_at()
119+
RETURNS TRIGGER AS $
120+
BEGIN
121+
NEW.updated_at = NOW();
122+
RETURN NEW;
123+
END;
124+
$ LANGUAGE plpgsql;
125+
126+
DROP TRIGGER IF EXISTS trg_batch_item_updated_at ON batch_items;
127+
CREATE TRIGGER trg_batch_item_updated_at
128+
BEFORE UPDATE ON batch_items
129+
FOR EACH ROW
130+
EXECUTE FUNCTION update_batch_item_updated_at();

‎backend/migrations/20260527000001_session_management.sql‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
-- User Session Management
2+
-- Creates tables for session tracking, device fingerprinting, and security monitoring.
3+
4+
-- -------------------------------------------------------------------------
5+
-- user_sessions
6+
-- -------------------------------------------------------------------------
7+
CREATE TABLE IF NOT EXISTS user_sessions (
8+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
9+
user_id VARCHAR(255) NOT NULL,
10+
session_token VARCHAR(512) UNIQUE NOT NULL,
11+
device_id VARCHAR(255) NOT NULL,
12+
device_fingerprint VARCHAR(512) NOT NULL,
13+
ip_address VARCHAR(45) NOT NULL,
14+
user_agent TEXT NOT NULL,
15+
status VARCHAR(50) NOT NULL DEFAULT 'active',
16+
last_activity TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
17+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
18+
expires_at TIMESTAMP WITH TIME ZONE NOT NULL,
19+
20+
CONSTRAINT fk_session_user
21+
FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE,
22+
23+
CONSTRAINT chk_session_status CHECK (
24+
status IN ('active', 'suspended', 'revoked', 'expired')
25+
)
26+
);
27+
28+
-- -------------------------------------------------------------------------
29+
-- session_activity_log
30+
-- -------------------------------------------------------------------------
31+
CREATE TABLE IF NOT EXISTS session_activity_log (
32+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
33+
session_id UUID NOT NULL,
34+
user_id VARCHAR(255) NOT NULL,
35+
activity_type VARCHAR(100) NOT NULL,
36+
endpoint VARCHAR(255),
37+
method VARCHAR(10),
38+
status_code INTEGER,
39+
ip_address VARCHAR(45),
40+
user_agent TEXT,
41+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
42+
43+
CONSTRAINT fk_activity_session
44+
FOREIGN KEY (session_id) REFERENCES user_sessions(id) ON DELETE CASCADE,
45+
46+
CONSTRAINT fk_activity_user
47+
FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
48+
);
49+
50+
-- -------------------------------------------------------------------------
51+
-- session_security_events
52+
-- -------------------------------------------------------------------------
53+
CREATE TABLE IF NOT EXISTS session_security_events (
54+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
55+
user_id VARCHAR(255) NOT NULL,
56+
session_id UUID,
57+
event_type VARCHAR(100) NOT NULL,
58+
severity VARCHAR(50) NOT NULL DEFAULT 'medium',
59+
description TEXT NOT NULL,
60+
ip_address VARCHAR(45),
61+
device_id VARCHAR(255),
62+
action_taken VARCHAR(100),
63+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
64+
65+
CONSTRAINT fk_security_event_user
66+
FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE,
67+
68+
CONSTRAINT fk_security_event_session
69+
FOREIGN KEY (session_id) REFERENCES user_sessions(id) ON DELETE SET NULL,
70+
71+
CONSTRAINT chk_event_type CHECK (
72+
event_type IN ('login', 'logout', 'failed_login', 'suspicious_activity', 'device_change', 'location_change', 'concurrent_session')
73+
),
74+
75+
CONSTRAINT chk_severity CHECK (
76+
severity IN ('low', 'medium', 'high', 'critical')
77+
)
78+
);
79+
80+
-- -------------------------------------------------------------------------
81+
-- Indexes
82+
-- -------------------------------------------------------------------------
83+
CREATE INDEX IF NOT EXISTS idx_sessions_user_id
84+
ON user_sessions(user_id);
85+
86+
CREATE INDEX IF NOT EXISTS idx_sessions_status
87+
ON user_sessions(status);
88+
89+
CREATE INDEX IF NOT EXISTS idx_sessions_expires_at
90+
ON user_sessions(expires_at);
91+
92+
CREATE INDEX IF NOT EXISTS idx_sessions_device_id
93+
ON user_sessions(device_id);
94+
95+
CREATE INDEX IF NOT EXISTS idx_activity_session_id
96+
ON session_activity_log(session_id);
97+
98+
CREATE INDEX IF NOT EXISTS idx_activity_user_id
99+
ON session_activity_log(user_id);
100+
101+
CREATE INDEX IF NOT EXISTS idx_activity_created_at
102+
ON session_activity_log(created_at DESC);
103+
104+
CREATE INDEX IF NOT EXISTS idx_security_events_user_id
105+
ON session_security_events(user_id);
106+
107+
CREATE INDEX IF NOT EXISTS idx_security_events_session_id
108+
ON session_security_events(session_id);
109+
110+
CREATE INDEX IF NOT EXISTS idx_security_events_event_type
111+
ON session_security_events(event_type);
112+
113+
CREATE INDEX IF NOT EXISTS idx_security_events_created_at
114+
ON session_security_events(created_at DESC);
115+
116+
-- -------------------------------------------------------------------------
117+
-- Trigger: auto-update last_activity on user_sessions
118+
-- -------------------------------------------------------------------------
119+
CREATE OR REPLACE FUNCTION update_session_last_activity()
120+
RETURNS TRIGGER AS $
121+
BEGIN
122+
NEW.last_activity = NOW();
123+
RETURN NEW;
124+
END;
125+
$ LANGUAGE plpgsql;
126+
127+
DROP TRIGGER IF EXISTS trg_session_last_activity ON user_sessions;
128+
CREATE TRIGGER trg_session_last_activity
129+
BEFORE UPDATE ON user_sessions
130+
FOR EACH ROW
131+
EXECUTE FUNCTION update_session_last_activity();

‎backend/migrations/20260527000002_stellar_events_indexer.sql‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
-- Stellar Event Indexing
2+
-- Creates tables for tracking Stellar network events and indexer state.
3+
4+
-- -------------------------------------------------------------------------
5+
-- stellar_events
6+
-- -------------------------------------------------------------------------
7+
CREATE TABLE IF NOT EXISTS stellar_events (
8+
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
9+
event_type VARCHAR(100) NOT NULL,
10+
tx_hash VARCHAR(64) UNIQUE NOT NULL,
11+
ledger_sequence BIGINT NOT NULL,
12+
source_account VARCHAR(56) NOT NULL,
13+
destination_account VARCHAR(56),
14+
asset_code VARCHAR(12),
15+
amount BIGINT,
16+
status VARCHAR(50) NOT NULL DEFAULT 'pending',
17+
processed BOOLEAN NOT NULL DEFAULT false,
18+
raw_data TEXT,
19+
indexed_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
20+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
21+
22+
CONSTRAINT chk_event_status CHECK (
23+
status IN ('pending', 'confirmed', 'failed')
24+
)
25+
);
26+
27+
-- -------------------------------------------------------------------------
28+
-- indexer_state
29+
-- -------------------------------------------------------------------------
30+
CREATE TABLE IF NOT EXISTS indexer_state (
31+
key VARCHAR(255) PRIMARY KEY,
32+
value TEXT NOT NULL,
33+
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
34+
);
35+
36+
-- -------------------------------------------------------------------------
37+
-- Indexes
38+
-- -------------------------------------------------------------------------
39+
CREATE INDEX IF NOT EXISTS idx_stellar_events_tx_hash
40+
ON stellar_events(tx_hash);
41+
42+
CREATE INDEX IF NOT EXISTS idx_stellar_events_event_type
43+
ON stellar_events(event_type);
44+
45+
CREATE INDEX IF NOT EXISTS idx_stellar_events_source_account
46+
ON stellar_events(source_account);
47+
48+
CREATE INDEX IF NOT EXISTS idx_stellar_events_destination_account
49+
ON stellar_events(destination_account);
50+
51+
CREATE INDEX IF NOT EXISTS idx_stellar_events_status
52+
ON stellar_events(status);
53+
54+
CREATE INDEX IF NOT EXISTS idx_stellar_events_processed
55+
ON stellar_events(processed);
56+
57+
CREATE INDEX IF NOT EXISTS idx_stellar_events_ledger_sequence
58+
ON stellar_events(ledger_sequence DESC);
59+
60+
CREATE INDEX IF NOT EXISTS idx_stellar_events_created_at
61+
ON stellar_events(created_at DESC);

‎backend/src/app.rs‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ use tower_http::{cors::CorsLayer, trace::TraceLayer};
1010
use crate::{
1111
config::Config,
1212
http::{
13-
admin, audit, auth, disputes, files, health, identity, jobs, metrics as metrics_http,
13+
admin, audit, auth, batches, disputes, files, health, identity, jobs, metrics as metrics_http,
1414
notifications, payments, profiles, transfers, version as version_http, withdrawals,
1515
},
1616
job_worker::JobWorker,
@@ -132,6 +132,15 @@ pub async fn create_app(
132132
.route("/audit-logs/:id", get(audit::get_audit_log))
133133
.layer(middleware::from_fn(role_guard::admin_only()));
134134

135+
// -------------------- Batches --------------------
136+
let batch_routes = Router::new()
137+
.route("/batches", post(batches::create_batch))
138+
.route("/batches/:batch_id", get(batches::get_batch))
139+
.route("/batches/:batch_id/items", post(batches::add_payment_to_batch))
140+
.route("/batches/:batch_id/report", get(batches::get_batch_report))
141+
.route("/batches/:batch_id/process", post(batches::process_batch))
142+
.route("/batches/merchant/:merchant_id", get(batches::get_merchant_batches));
143+
135144
// -------------------- Disputes (v2 only) --------------------
136145
// Payment-scoped dispute routes
137146
let payment_dispute_routes = Router::new()
@@ -178,6 +187,7 @@ pub async fn create_app(
178187
.nest("/notifications", notification_routes)
179188
.nest("/profiles", profile_routes)
180189
.nest("/files", files_routes)
190+
.nest("/batches", batch_routes)
181191
.nest("/admin", admin_routes)
182192
.nest("/audit", audit_routes);
183193

0 commit comments

Comments
 (0)