Ingestion Pipeline

Scope: owns the capture path — listener, extraction, filtering, pipeline sequence, routing, replay.

The capture path turns a posted notification into a transaction (or a review-queue entry). It runs entirely off the main thread.

Components

Sequence

sequenceDiagram
    autonumber
    actor B as Bank app
    participant NMS as NotificationManagerService
    participant L as TransactionNotificationListener
    participant EX as NotificationExtractor
    participant P as CapturePipeline
    participant F as NotificationFilter
    participant PE as ParserEngine
    participant CL as CategoryClassifier
    participant DB as Room + SQLCipher
    participant TN as TransactionNotifier

    B->>NMS: post notification
    NMS->>L: onNotificationPosted(sbn)
    L->>EX: extractText(sbn)
    EX-->>L: sanitized text (digits masked)
    L->>P: onNotification(pkg, text, postTime)
    P->>P: key = NotificationKey.compute(...)
    alt session duplicate (seen in sessionSeen)
        P-->>L: return true (skipped)
    else
        alt isExcluded (promo/OTP)
            P-->>L: return true (dropped)
        else hasTransactionSignal == false
            alt isMaskedContent
                P->>DB: insert UnmatchedEntity(isMasked=true)
            end
            P-->>L: return true
        else
            P->>PE: match(pkg, text)  [rules from rules table]
            alt candidate == null
                P->>DB: insert UnmatchedEntity
            else candidate
                P->>CL: classify(merchant, amount, direction, pkg)
                CL-->>P: categoryId?
                P->>DB: insert TransactionEntity(notificationKey, isHidden=status!=COMPLETED, parseSource, rawJson?)
                Note over P,DB: unique index on notificationKey → duplicate returns -1, no-op
                Note over P: new row + notify_on_parse != "false"
                P->>TN: show(transaction, categoryName)
                TN->>NMS: post silent transaction-captured notification
            end
            P-->>L: return true
        end
    end

Listener reconnect & replay

On onListenerConnected() the listener replays notifications posted in the last 48h (REPLAY_WINDOW_MILLIS) via activeNotifications, feeding each through onNotificationPosted. This covers the gap when the listener was unbound or the process died. Replay is idempotent: the session key set and the DB unique index both swallow duplicates.

The listener never holds a wake lock and only wakes on posted notifications or on bind.

Routing decisions

The session key is computed from the raw sanitized text before the exclusion/signal gates, so a repeated masked or signal-less alert can't re-queue through session dedupe.

Captured location

When capture-location is enabled (opt-in; off by default), the listener attaches a device fix to each notification before the pipeline runs. With the "Allow all the time" grant (ADR-0034), LocationProvider.currentLocation() reuses a cached fix at most 2 min old and otherwise requests a live one-shot fix (framework requestSingleUpdate, 10 s timeout), falling back to the freshest cached fix (GPS, network, passive; ≤24 h old) when the request times out. A background caller holding only while-in-use location permission is app-op-blocked by the framework — every read silently returns null — which is why capture requires the background grant. The pipeline stores the fix on the transaction or unmatched row when present (ADR-0021, ADR-0026, ADR-0034).

Cross-source dedupe

Google Wallet card charges also arrive as a notification from the issuing bank app. TransactionRepository.resolveCrossSourceDuplicate treats the bank app as authoritative: a Wallet alert is suppressed when a completed bank transaction with the same amount + currency + direction exists within ±90 s, and a bank alert replaces an existing Wallet-only row (ADR-0020). Only COMPLETED candidates participate; pending/failed alerts never suppress a real charge.

Threading & failure handling

• All work is suspend and runs on Dispatchers.IO; writes are never on the main thread.
• Every failure boundary is logged and swallowed (Log.w) so the system service or the receiver never crashes: text extraction, parsing, classification, per-notification replay.
• The session dedupe set is bounded (MAX_SESSION_KEYS = 512, cleared when full) and guarded by synchronized; it is a small LinkedHashSet<String>.

Related

• Parser engine
• Classification
• Capture gating & replay ADR
• Confidence gate ADR
• Cross-source duplicate suppression ADR
• Captured-location data ADR
• Transaction notification & parse-source ADR