pispas_modules/

paytef.rs

use crate::service::{Service, WebSocketWrite};
use async_trait::async_trait;
use futures_util::SinkExt;
use serde_json::{json, Value};
use std::any::Any;
use std::sync::Arc;
use tokio_tungstenite::tungstenite::protocol::Message;
use easy_trace::prelude::{error, info, warn};

const PAYTEF_VERSION: &str = "1.0.0";
const POLL_INTERVAL_MS: u64 = 1000;

/// Hard ceiling on the start->result poll loop. The terminal must reach a
/// terminal status (finished/cancelled/error) within this window or we
/// give up, fire a best-effort cancel, and surface a deterministic error
/// to the caller. 5 minutes is generous for a card present flow (PIN
/// entry, contactless retry, signature) but short enough that a hung
/// terminal does not pin the WS task forever.
const TRANSACTION_TIMEOUT_SECS: u64 = 300;

/// Single key in PersistenceService; value is a JSON object mapping
/// transaction reference -> cached sale entry. One key (as opposed to
/// N `paytef_sale:{reference}` keys) because PersistenceService
/// serializes+encrypts+writes the whole store on every set_value, so
/// grouping costs the same and makes pruning a single read-modify-write.
const CACHE_KEY: &str = "paytef_sales";

/// Max age for a cached sale to be eligible for a referenced refund.
/// 24h covers end-of-day closings and late-night venues that keep the
/// till open past midnight — later refunds go through admin and don't
/// need the paytef operation link.
const CACHE_TTL_SECS: u64 = 24 * 3600;

pub struct PaytefService {
    /// Shared with the rest of the service. We reuse it as a cache for
    /// completed paytef sales so that a later refund on the same ticket
    /// can reference the original operation (paytef "referenced refund"
    /// flow). All paytef sales live under a single key `paytef_sales`
    /// (see CACHE_KEY) as a map keyed by transaction reference (the web
    /// client passes ticket.orderno today). Encrypted + persisted to
    /// disk by PersistenceService, so a service restart does not force
    /// us into a blind/unreferenced refund.
    persistence: Option<Arc<crate::persistence::PersistenceService>>,
}

impl PaytefService {
    pub fn new(persistence: Option<Arc<crate::persistence::PersistenceService>>) -> Self {
        info!("PaytefService initialized");
        Self { persistence }
    }

    fn now_secs() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0)
    }

    /// Remove entries whose `created_at` is older than `ttl` relative to
    /// `now`. Returns the number of entries dropped so we can surface it
    /// in the logs. Also drops entries without a parseable `created_at`
    /// (treated as infinitely old) — leftovers from an earlier format or
    /// hand-edited state shouldn't keep the cache growing.
    fn prune_expired(map: &mut serde_json::Map<String, Value>, now: u64, ttl: u64) -> usize {
        let before = map.len();
        map.retain(|_, v| {
            let created = v.get("created_at").and_then(|c| c.as_u64()).unwrap_or(0);
            created > 0 && now.saturating_sub(created) < ttl
        });
        before - map.len()
    }

    /// Read the sales map from persistence. Returns an empty map when
    /// there is no persistence, the key is missing, or the value is the
    /// wrong shape. Callers are expected to mutate and write back via
    /// save_sales; there is no separate "get mutable ref" because
    /// PersistenceService stores serde Values, not mutable handles.
    async fn load_sales(&self) -> serde_json::Map<String, Value> {
        let persistence = match &self.persistence {
            Some(p) => p,
            None => return serde_json::Map::new(),
        };
        let cache = persistence.cache.lock().await;
        cache.get(CACHE_KEY)
            .and_then(|v| v.as_object())
            .cloned()
            .unwrap_or_default()
    }

    async fn save_sales(&self, map: serde_json::Map<String, Value>) {
        let persistence = match &self.persistence {
            Some(p) => p,
            None => return,
        };
        if let Err(e) = persistence.set_value(CACHE_KEY.to_string(), Value::Object(map)).await {
            warn!("Failed to persist paytef sales cache: {}", e);
        }
    }

    fn base_url(host: &str, port: u16) -> String {
        format!("http://{}:{}", host, port)
    }

    async fn send_ws_message(write: &WebSocketWrite, message: Value) {
        if let Some(ws_lock) = write {
            let serialized = Message::Text(message.to_string());
            let mut ws = ws_lock.write().await;
            if let Err(e) = ws.send(serialized).await {
                error!("Failed to send WebSocket message: {}", e);
            }
        }
    }

    async fn http_post(url: &str, body: &Value) -> Result<Value, String> {
        // Paytef terminal firmware speaks only HTTP/1.1 with Connection:
        // Keep-Alive. reqwest defaults to HTTP/2 auto-upgrade which fails
        // silently against HTTP/1-only servers (you get a terse "error
        // sending request" in ~3 ms with no detail). Forcing http1_only
        // avoids the upgrade dance.
        let client = reqwest::Client::builder()
            .http1_only()
            .timeout(std::time::Duration::from_secs(10))
            .build()
            .map_err(|e| format!("HTTP client build failed: {}", e))?;

        let response = client
            .post(url)
            .json(body)
            .send()
            .await
            .map_err(|e| {
                // Extract the root cause — reqwest's Display drops the
                // chain and leaves a useless "error sending request".
                let mut cause = format!("{}", e);
                let mut src: Option<&(dyn std::error::Error + 'static)> =
                    std::error::Error::source(&e);
                while let Some(s) = src {
                    cause.push_str(&format!(" | caused by: {}", s));
                    src = s.source();
                }
                format!("HTTP request failed: {}", cause)
            })?;

        let status = response.status();
        let text = response
            .text()
            .await
            .unwrap_or_else(|_| "{}".to_string());

        if !status.is_success() {
            return Err(format!("HTTP {}: {}", status, text));
        }

        serde_json::from_str(&text)
            .map_err(|e| format!("JSON parse error: {} - body: {}", e, text))
    }

    /// Generic transaction runner — handles start → poll → result for any
    /// opType paytef supports (sale, refund, void, preauthorization,
    /// completion, installments). For sale we don't pass a
    /// previous_operation_number; for refund/void we optionally do to get
    /// a "referenced" operation linked to the original.
    async fn transaction(
        op_type: &str,
        host: &str,
        port: u16,
        amount: i64,
        pinpad: &str,
        reference: &str,
        previous_operation_number: Option<i64>,
        write: &WebSocketWrite,
    ) -> Result<Value, String> {
        let base = Self::base_url(host, port);

        // 1. Start transaction
        info!(op_type = op_type, host = host, port = port, amount = amount, "Starting Paytef transaction");
        let mut start_body = json!({
            "language": "es",
            "pinpad": pinpad,
            "executeOptions": { "method": "polling" },
            "opType": op_type,
            "requestedAmount": amount,
            "createReceipt": false,
            "showResultSeconds": 5,
            "transactionReference": reference
        });

        if let Some(prev) = previous_operation_number {
            start_body["operationDetails"] = json!({
                "previousOperationNumber": prev
            });
        }

        let start_resp = Self::http_post(
            &format!("{}/transaction/start", base),
            &start_body,
        ).await?;

        let started = start_resp
            .get("info")
            .and_then(|i| i.get("started"))
            .and_then(|s| s.as_bool())
            .unwrap_or(false);

        if !started {
            let msg = start_resp
                .get("info")
                .and_then(|i| i.get("message"))
                .and_then(|m| m.as_str())
                .unwrap_or("Unknown error");
            return Err(format!("Failed to start transaction: {}", msg));
        }

        let session_id = start_resp
            .get("info")
            .and_then(|i| i.get("sessionID"))
            .and_then(|s| s.as_str())
            .unwrap_or("")
            .to_string();

        info!(session_id = %session_id, "Paytef transaction started");

        // Send initial progress
        Self::send_ws_message(write, json!({
            "status": "started",
            "opType": op_type,
            "sessionID": session_id
        })).await;

        // 2. Poll loop
        //
        // Bounded by TRANSACTION_TIMEOUT_SECS so a terminal that never
        // reaches a terminal status (firmware bug, network partition mid-
        // session, customer walking away) cannot hang the WS task
        // forever. On timeout we fire a best-effort `cancel` so the
        // terminal is not left in a busy state and surface a determinate
        // error to the caller.
        let poll_body = json!({ "pinpad": pinpad });
        let deadline = tokio::time::Instant::now()
            + std::time::Duration::from_secs(TRANSACTION_TIMEOUT_SECS);
        loop {
            if tokio::time::Instant::now() >= deadline {
                warn!(
                    op_type = op_type,
                    session_id = %session_id,
                    timeout_secs = TRANSACTION_TIMEOUT_SECS,
                    "Paytef transaction poll timed out, cancelling"
                );
                Self::send_ws_message(write, json!({
                    "status": "progress",
                    "opType": op_type,
                    "transactionStatus": "timeout",
                    "sessionID": session_id
                })).await;
                // Best-effort cancel — log and ignore failures; we are
                // already returning an error.
                if let Err(e) = Self::cancel(host, port, pinpad).await {
                    warn!(error = %e, "Cancel after timeout failed");
                }
                return Err(format!(
                    "Transaction timed out after {}s without reaching a terminal status",
                    TRANSACTION_TIMEOUT_SECS
                ));
            }

            tokio::time::sleep(std::time::Duration::from_millis(POLL_INTERVAL_MS)).await;

            let poll_resp = Self::http_post(
                &format!("{}/transaction/poll", base),
                &poll_body,
            ).await?;

            let poll_info = poll_resp.get("info").cloned().unwrap_or(json!({}));
            let tx_status = poll_info.get("transactionStatus")
                .and_then(|s| s.as_str())
                .unwrap_or("");
            let card_status = poll_info.get("cardStatus")
                .and_then(|s| s.as_str())
                .unwrap_or("");

            // Send progress update
            Self::send_ws_message(write, json!({
                "status": "progress",
                "opType": op_type,
                "transactionStatus": tx_status,
                "cardStatus": card_status,
                "sessionID": session_id
            })).await;

            // Check if result is available in poll response
            if let Some(result) = poll_resp.get("result") {
                if !result.is_null() {
                    break;
                }
            }

            // Terminal states
            if tx_status == "finished" || tx_status == "cancelled" || tx_status == "error" {
                break;
            }
        }

        // 3. Get result
        let result_resp = Self::http_post(
            &format!("{}/transaction/result", base),
            &poll_body,
        ).await?;

        let result = result_resp.get("result").cloned().unwrap_or(Value::Null);

        if result.is_null() {
            return Err("Transaction cancelled or no result".to_string());
        }

        let approved = result.get("approved")
            .and_then(|a| a.as_bool())
            .unwrap_or(false);

        let response = json!({
            "status": if approved { "approved" } else { "denied" },
            "opType": op_type,
            "approved": approved,
            "resultCode": result.get("resultCode"),
            "resultText": result.get("resultText"),
            "authorisationCode": result.get("authorisationCode"),
            "cardInformation": result.get("cardInformation"),
            "requestedAmount": result.get("requestedAmount"),
            "paytefOperationNumber": result.get("paytefOperationNumber"),
            "sessionID": session_id
        });

        info!(
            op_type = op_type,
            approved = approved,
            session_id = %session_id,
            paytef_op = ?result.get("paytefOperationNumber"),
            "Paytef transaction completed"
        );

        Ok(response)
    }

    async fn cancel(host: &str, port: u16, pinpad: &str) -> Result<Value, String> {
        let base = Self::base_url(host, port);
        let body = json!({ "pinpad": pinpad });
        let resp = Self::http_post(&format!("{}/pinpad/cancel", base), &body).await?;
        Ok(resp)
    }

    async fn status(host: &str, port: u16) -> Result<Value, String> {
        let base = Self::base_url(host, port);
        let body = json!({ "pinpad": "*" });
        let resp = Self::http_post(&format!("{}/pinpad/status", base), &body).await?;
        Ok(resp)
    }

    async fn print(host: &str, port: u16, content: &str, pinpad: &str) -> Result<Value, String> {
        let base = Self::base_url(host, port);
        let body = json!({
            "pinpad": pinpad,
            "content": content,
            "contentType": "html",
            "usePrinter": true,
            "generateImage": false
        });
        let resp = Self::http_post(&format!("{}/printer/print", base), &body).await?;
        Ok(resp)
    }

    /// Pull a cached sale by reference, if any. Returns None when the
    /// entry is missing or older than CACHE_TTL_SECS (stale entries get
    /// reaped on the next write — we don't need a timer). No-op when
    /// persistence is not available or reference is empty.
    async fn lookup_sale(&self, reference: &str) -> Option<Value> {
        if reference.is_empty() {
            return None;
        }
        let map = self.load_sales().await;
        let entry = map.get(reference)?.clone();
        let created = entry.get("created_at").and_then(|c| c.as_u64()).unwrap_or(0);
        if created == 0 || Self::now_secs().saturating_sub(created) >= CACHE_TTL_SECS {
            return None;
        }
        Some(entry)
    }

    /// Insert a sale and prune everything older than CACHE_TTL_SECS in
    /// the same write. Runs on every approved sale, which means the
    /// cache stays bounded even on devices that never reboot.
    async fn remember_sale(
        &self,
        reference: &str,
        paytef_operation_number: i64,
        amount: i64,
        host: &str,
        port: u16,
        pinpad: &str,
    ) {
        if reference.is_empty() {
            return;
        }
        if self.persistence.is_none() {
            warn!("Paytef sale not cached — no persistence service available");
            return;
        }
        let now = Self::now_secs();
        let mut map = self.load_sales().await;
        map.insert(reference.to_string(), json!({
            "paytef_operation_number": paytef_operation_number,
            "amount": amount,
            "host": host,
            "port": port,
            "pinpad": pinpad,
            "created_at": now,
        }));
        let pruned = Self::prune_expired(&mut map, now, CACHE_TTL_SECS);
        if pruned > 0 {
            info!("Paytef cache: pruned {} expired sale(s) on write", pruned);
        }
        self.save_sales(map).await;
    }

    async fn forget_sale(&self, reference: &str) {
        if reference.is_empty() {
            return;
        }
        if self.persistence.is_none() {
            return;
        }
        let mut map = self.load_sales().await;
        if map.remove(reference).is_some() {
            self.save_sales(map).await;
        }
    }
}

#[async_trait]
impl Service for PaytefService {
    async fn run(&self, action: Value, write: WebSocketWrite) -> (i32, String) {
        // ACTION arrives either as a JSON object (used by internal tests) or
        // as a JSON-encoded string. The web client's WS envelope always wraps
        // ACTION as a string for every service (BASE/CHECK sends "CHECK",
        // paytef sends the stringified command payload), so the string form
        // is what real traffic looks like. Accept both.
        let owned: Value = match action.get("ACTION") {
            Some(v) if v.is_object() => v.clone(),
            Some(v) if v.is_string() => match serde_json::from_str::<Value>(v.as_str().unwrap_or("")) {
                Ok(parsed) if parsed.is_object() => parsed,
                _ => {
                    error!("Invalid action format: ACTION string is not a JSON object");
                    return (1, "Invalid action format: ACTION string is not a JSON object".to_string());
                }
            },
            _ => {
                error!("Invalid action format: missing 'ACTION'");
                return (1, "Invalid action format: missing 'ACTION'".to_string());
            }
        };
        let action_data = owned.as_object().expect("owned guaranteed to be object");

        let command = action_data
            .get("command")
            .and_then(|c| c.as_str())
            .unwrap_or("");

        let host = action_data
            .get("host")
            .and_then(|h| h.as_str())
            .unwrap_or("127.0.0.1");

        let port = action_data
            .get("port")
            .and_then(|p| p.as_u64())
            .unwrap_or(8887) as u16;

        let pinpad = action_data
            .get("pinpad")
            .and_then(|p| p.as_str())
            .unwrap_or("*");

        match command {
            "SALE" => {
                let amount = action_data
                    .get("amount")
                    .and_then(|a| a.as_i64())
                    .unwrap_or(0);

                let reference = action_data
                    .get("reference")
                    .and_then(|r| r.as_str())
                    .unwrap_or("");

                if amount <= 0 {
                    return (1, "Invalid amount: must be > 0".to_string());
                }

                match Self::transaction("sale", host, port, amount, pinpad, reference, None, &write).await {
                    Ok(result) => {
                        // Cache the sale for a future referenced refund — only
                        // when approved AND we have an op number to reference.
                        let approved = result.get("approved").and_then(|a| a.as_bool()).unwrap_or(false);
                        let op_number = result.get("paytefOperationNumber").and_then(|n| n.as_i64());
                        if approved {
                            if let Some(op) = op_number {
                                self.remember_sale(reference, op, amount, host, port, pinpad).await;
                            }
                            else {
                                warn!(reference = reference, "Paytef sale approved but no paytefOperationNumber in result — skipping cache (referenced refund won't be possible)");
                            }
                        }
                        (0, result.to_string())
                    }
                    Err(e) => {
                        error!(error = %e, "Paytef SALE failed");
                        (1, json!({ "status": "error", "error": e }).to_string())
                    }
                }
            }

            "REFUND" => {
                let reference = action_data
                    .get("reference")
                    .and_then(|r| r.as_str())
                    .unwrap_or("");

                // Amount is optional: if we have a cached sale for this
                // reference and no amount was passed, refund the whole thing.
                let explicit_amount = action_data
                    .get("amount")
                    .and_then(|a| a.as_i64());

                let cached = self.lookup_sale(reference).await;

                // Extract fields from cached JSON value (remember_sale shape)
                let cached_amount = cached.as_ref().and_then(|v| v.get("amount")).and_then(|a| a.as_i64());
                let cached_op = cached.as_ref().and_then(|v| v.get("paytef_operation_number")).and_then(|a| a.as_i64());
                let cached_host = cached.as_ref().and_then(|v| v.get("host")).and_then(|a| a.as_str()).map(|s| s.to_string());
                let cached_port = cached.as_ref().and_then(|v| v.get("port")).and_then(|a| a.as_u64()).map(|p| p as u16);
                let cached_pinpad = cached.as_ref().and_then(|v| v.get("pinpad")).and_then(|a| a.as_str()).map(|s| s.to_string());

                let (amount, prev_op, eff_host, eff_port, eff_pinpad) = match (cached.is_some(), explicit_amount) {
                    // Full refund via cache (no explicit amount)
                    (true, None) => (
                        cached_amount.unwrap_or(0),
                        cached_op,
                        cached_host.unwrap_or_else(|| host.to_string()),
                        cached_port.unwrap_or(port),
                        cached_pinpad.unwrap_or_else(|| pinpad.to_string()),
                    ),
                    // Partial (or explicit full) refund with cached reference
                    (true, Some(amt)) => (
                        amt,
                        cached_op,
                        cached_host.unwrap_or_else(|| host.to_string()),
                        cached_port.unwrap_or(port),
                        cached_pinpad.unwrap_or_else(|| pinpad.to_string()),
                    ),
                    // Unreferenced refund — no cache hit. Paytef will accept
                    // it but the refund won't be linked to the original sale.
                    (false, Some(amt)) => (
                        amt,
                        None,
                        host.to_string(),
                        port,
                        pinpad.to_string(),
                    ),
                    // Nothing to refund
                    (false, None) => {
                        return (1, "Missing amount and no cached sale for reference".to_string());
                    }
                };

                if amount <= 0 {
                    return (1, "Invalid amount: must be > 0".to_string());
                }

                match Self::transaction("refund", &eff_host, eff_port, amount, &eff_pinpad, reference, prev_op, &write).await {
                    Ok(result) => {
                        let approved = result.get("approved").and_then(|a| a.as_bool()).unwrap_or(false);
                        if approved && cached.is_some() {
                            // For a full refund, drop the cache entry so the
                            // ticket can't be refunded twice. For partial we
                            // keep it — the user may need further refunds.
                            let full = cached_amount == Some(amount);
                            if full {
                                self.forget_sale(reference).await;
                            }
                        }
                        (0, result.to_string())
                    }
                    Err(e) => {
                        error!(error = %e, "Paytef REFUND failed");
                        (1, json!({ "status": "error", "error": e }).to_string())
                    }
                }
            }

            "CANCEL" => {
                match Self::cancel(host, port, pinpad).await {
                    Ok(result) => (0, result.to_string()),
                    Err(e) => {
                        error!(error = %e, "Paytef CANCEL failed");
                        (1, json!({ "status": "error", "error": e }).to_string())
                    }
                }
            }

            "STATUS" => {
                match Self::status(host, port).await {
                    Ok(result) => (0, result.to_string()),
                    Err(e) => {
                        error!(error = %e, "Paytef STATUS failed");
                        (1, json!({ "status": "error", "error": e }).to_string())
                    }
                }
            }

            "PRINT" => {
                let content = action_data
                    .get("content")
                    .and_then(|c| c.as_str())
                    .unwrap_or("");

                if content.is_empty() {
                    return (1, "Missing print content".to_string());
                }

                match Self::print(host, port, content, pinpad).await {
                    Ok(result) => (0, result.to_string()),
                    Err(e) => {
                        error!(error = %e, "Paytef PRINT failed");
                        (1, json!({ "status": "error", "error": e }).to_string())
                    }
                }
            }

            _ => {
                error!(command = command, "Unknown Paytef command");
                (1, format!("Unknown command: {}", command))
            }
        }
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn stop_service(&self) {
        info!("PaytefService stopped.");
    }

    fn get_version(&self) -> String {
        PAYTEF_VERSION.to_string()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_base_url() {
        assert_eq!(
            PaytefService::base_url("192.168.1.178", 8887),
            "http://192.168.1.178:8887"
        );
    }

    #[test]
    fn test_base_url_localhost() {
        assert_eq!(
            PaytefService::base_url("127.0.0.1", 8887),
            "http://127.0.0.1:8887"
        );
    }

    #[tokio::test]
    async fn test_sale_invalid_amount() {
        let service = PaytefService::new(None);
        let action = json!({
            "ACTION": {
                "command": "SALE",
                "host": "127.0.0.1",
                "port": 8887,
                "amount": 0,
                "pinpad": "*"
            }
        });

        let (status, msg) = service.run(action, None).await;
        assert_eq!(status, 1);
        assert!(msg.contains("Invalid amount"));
    }

    #[tokio::test]
    async fn test_missing_action() {
        let service = PaytefService::new(None);
        let action = json!({ "no_action": true });

        let (status, msg) = service.run(action, None).await;
        assert_eq!(status, 1);
        assert!(msg.contains("missing 'ACTION'"));
    }

    #[tokio::test]
    async fn test_unknown_command() {
        let service = PaytefService::new(None);
        let action = json!({
            "ACTION": {
                "command": "UNKNOWN_CMD",
                "host": "127.0.0.1",
                "port": 8887
            }
        });

        let (status, msg) = service.run(action, None).await;
        assert_eq!(status, 1);
        assert!(msg.contains("Unknown command"));
    }

    #[tokio::test]
    async fn test_refund_no_reference_no_amount() {
        let service = PaytefService::new(None);
        let action = json!({
            "ACTION": {
                "command": "REFUND",
                "host": "127.0.0.1",
                "port": 8887
            }
        });

        let (status, msg) = service.run(action, None).await;
        assert_eq!(status, 1);
        assert!(msg.contains("Missing amount"));
    }

    #[test]
    fn test_prune_expired_drops_old_entries() {
        // now = 100000, ttl = 3600 → entries with created_at < 96400 drop
        let mut map = serde_json::Map::new();
        map.insert("fresh".into(), json!({ "created_at": 99000_u64 }));
        map.insert("expired".into(), json!({ "created_at": 90000_u64 }));
        map.insert("no_timestamp".into(), json!({}));
        map.insert("zero".into(), json!({ "created_at": 0_u64 }));

        let dropped = PaytefService::prune_expired(&mut map, 100000, 3600);
        assert_eq!(dropped, 3);
        assert!(map.contains_key("fresh"));
        assert!(!map.contains_key("expired"));
        assert!(!map.contains_key("no_timestamp"));
        assert!(!map.contains_key("zero"));
    }

    #[test]
    fn test_prune_expired_keeps_everything_within_ttl() {
        let mut map = serde_json::Map::new();
        map.insert("a".into(), json!({ "created_at": 1000_u64 }));
        map.insert("b".into(), json!({ "created_at": 2000_u64 }));

        let dropped = PaytefService::prune_expired(&mut map, 2500, 3600);
        assert_eq!(dropped, 0);
        assert_eq!(map.len(), 2);
    }
}