feat(sync): add Composio stripe memory-sync pipeline - #107
Conversation
Implement StripeSyncPipeline (SyncPipeline + IncrementalSource) for the Composio `stripe` toolkit, modeled on the document-shaped Linear pipeline. Fetches charges via STRIPE_LIST_ALL_CHARGES with Stripe cursor pagination (`starting_after` + `has_more`). Uses the shared `document()` helper so the upsert key is the stable object id (`stripe:<charge_id>`) with `metadata.taint = "external_sync"` — never a per-run cursor, avoiding the duplicate-charge bug class fixed by tinyhumansai/openhuman#4953. No item-level financial data is logged. Registered through providers/mod.rs, composio/mod.rs, and sync/mod.rs alongside the six existing pipelines. Adds unit tests for toolkit()/action(), extract_page (charge extraction + cursor + has_more stop), stable dedupe key, and a stable document_id from document(). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
📝 WalkthroughWalkthroughAdds a Stripe Composio incremental sync pipeline that paginates charges, deduplicates records, creates stable ChangesStripe synchronization
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related issues
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/memory/sync/composio/providers/stripe.rs`:
- Around line 169-265: The tests are embedded in the implementation file instead
of a sibling test module. Move the entire #[cfg(test)] mod tests block,
including pipeline, sample_payload, and all test functions, into a sibling
stripe_tests.rs file; import the implementation symbols there as needed and
remove the embedded test module from stripe.rs.
- Line 27: Update the ACTION_LIST_CHARGES constant to use the Composio Stripe
charges slug STRIPE_LIST_CHARGES instead of STRIPE_LIST_ALL_CHARGES, preserving
the existing action lookup flow.
- Around line 46-50: Update with_limits so page_size is clamped to a maximum of
100 while retaining the existing minimum of 1, ensuring values passed to Stripe
remain within the accepted range. Leave max_pages handling unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 61ec22c4-33da-42c0-b985-64e1467e40a4
📒 Files selected for processing (4)
src/memory/sync/composio/mod.rssrc/memory/sync/composio/providers/mod.rssrc/memory/sync/composio/providers/stripe.rssrc/memory/sync/mod.rs
| SkillDocument, SyncContext, SyncOutcome, SyncPipeline, SyncPipelineKind, | ||
| }; | ||
|
|
||
| const ACTION_LIST_CHARGES: &str = "STRIPE_LIST_ALL_CHARGES"; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n 'STRIPE_LIST' --type=rust
rg -n 'STRIPE_LIST' -g '*.json' -g '*.yaml' -g '*.yml' -g '*.toml'Repository: tinyhumansai/tinycortex
Length of output: 230
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== relevant files ==\n'
git ls-files 'src/memory/sync/composio/providers/stripe.rs' 'src/**/*.rs' '*.json' '*.yaml' '*.yml' '*.toml' | sed -n '1,200p'
printf '\n== stripe provider ==\n'
sed -n '1,220p' src/memory/sync/composio/providers/stripe.rs
printf '\n== search composio/stripe slugs ==\n'
rg -n 'STRIPE_LIST_|composio|charges' src --glob '!**/target/**'
printf '\n== search repo-wide for exact slug and nearby variants ==\n'
rg -n 'STRIPE_LIST_ALL_CHARGES|STRIPE_LIST_CHARGES|LIST_ALL_CHARGES|LIST_CHARGES' .
printf '\n== file sizes for likely context ==\n'
wc -l src/memory/sync/composio/providers/stripe.rsRepository: tinyhumansai/tinycortex
Length of output: 23065
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- stripe provider outline ---'
ast-grep outline src/memory/sync/composio/providers/stripe.rs --view expanded || true
echo
echo '--- stripe provider excerpt ---'
sed -n '1,140p' src/memory/sync/composio/providers/stripe.rs
echo
echo '--- exact slug searches ---'
rg -n --fixed-strings 'STRIPE_LIST_ALL_CHARGES' src .
rg -n --fixed-strings 'STRIPE_LIST_CHARGES' src .
rg -n --fixed-strings 'STRIPE_LIST_' src .
echo
echo '--- composio-related files ---'
git ls-files | rg 'composio|stripe|composio.*(json|ya?ml|toml)$|.*(json|ya?ml|toml)$'Repository: tinyhumansai/tinycortex
Length of output: 6768
🌐 Web query:
Composio Stripe charges listing action slug STRIPE_LIST_CHARGES STRIPE_LIST_ALL_CHARGES
💡 Result:
In the Composio Stripe integration, the correct action slug for listing charges is STRIPE_LIST_CHARGES [1][2]. There is no standard action slug named STRIPE_LIST_ALL_CHARGES in the current Composio Stripe toolkit documentation [1][2]. The STRIPE_LIST_CHARGES action is designed to retrieve a list of Stripe charges and supports filtering and pagination [1][3]. When using this action, it is recommended to use valid cursor IDs from previous responses for pagination, as charges are typically returned in reverse chronological order [1][4]. Common parameters used with STRIPE_LIST_CHARGES include filters such as 'customer' and 'limit' [2][5].
Citations:
- 1: https://composio.dev/toolkits/stripe
- 2: https://playbooks.com/skills/composiohq/awesome-claude-skills/stripe-automation
- 3: https://composio.dev/toolkits/stripe/framework/autogen
- 4: https://composio.dev/toolkits/stripe/framework/google-adk
- 5: https://github.com/davepoon/buildwithclaude/blob/main/plugins/all-skills/skills/stripe-automation/SKILL.md
Use the Composio Stripe charges slug. STRIPE_LIST_ALL_CHARGES should be STRIPE_LIST_CHARGES; otherwise the action lookup fails and Stripe syncs stop working.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memory/sync/composio/providers/stripe.rs` at line 27, Update the
ACTION_LIST_CHARGES constant to use the Composio Stripe charges slug
STRIPE_LIST_CHARGES instead of STRIPE_LIST_ALL_CHARGES, preserving the existing
action lookup flow.
| pub fn with_limits(mut self, max_pages: usize, page_size: usize) -> Self { | ||
| self.max_pages = max_pages.max(1); | ||
| self.page_size = page_size.max(1); | ||
| self | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== stripe provider file ==\n'
sed -n '1,220p' src/memory/sync/composio/providers/stripe.rs
printf '\n== search for page_size usage ==\n'
rg -n "page_size|max_pages|with_limits|limit" src/memory/sync/composio/providers src/memory/sync -g '!**/target/**'Repository: tinyhumansai/tinycortex
Length of output: 19698
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Focus on the target file and its immediate context.
nl -ba src/memory/sync/composio/providers/stripe.rs | sed -n '1,220p'
printf '\n== page_size references in stripe provider subtree ==\n'
rg -n "page_size|max_pages|with_limits|limit" src/memory/sync/composio/providers/stripe.rs src/memory/sync/composio/providers -g '!**/target/**'Repository: tinyhumansai/tinycortex
Length of output: 201
🌐 Web query:
Stripe API list pagination limit parameter maximum 100 charges list documentation
💡 Result:
For the Stripe API, the limit parameter used in list endpoints, including the charges list, has a maximum allowed value of 100 [1][2][3]. The value must be an integer between 1 and 100, with a default value of 10 if no limit is specified [1][4][5]. To retrieve more than 100 objects, you should use cursor-based pagination with the starting_after or ending_before parameters [2][3].
Citations:
- 1: https://docs.stripe.com/api/charges/list
- 2: https://docs.stripe.com/api/pagination
- 3: https://docs.stripe.com/api/pagination?api-version=2026-04-22.dahlia
- 4: https://docs.stripe.com/api/charges/list?lang=node
- 5: https://docs.stripe.com/api/pagination?api-version=2024-11-20.acacia
Clamp page_size to 100. Stripe list endpoints reject limit values above 100, so larger values passed through with_limits will make every page request fail.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memory/sync/composio/providers/stripe.rs` around lines 46 - 50, Update
with_limits so page_size is clamped to a maximum of 100 while retaining the
existing minimum of 1, ensuring values passed to Stripe remain within the
accepted range. Leave max_pages handling unchanged.
| #[cfg(test)] | ||
| mod tests { | ||
| use super::*; | ||
| use crate::memory::config::{ComposioMode, ComposioSyncConfig}; | ||
|
|
||
| fn pipeline() -> StripeSyncPipeline { | ||
| let config = ComposioSyncConfig { | ||
| mode: ComposioMode::Direct, | ||
| base_url: "http://localhost".into(), | ||
| api_key: None, | ||
| bearer_token: None, | ||
| entity_id: None, | ||
| }; | ||
| StripeSyncPipeline::new(ComposioClient::new(config), "conn-test") | ||
| } | ||
|
|
||
| fn sample_payload() -> Value { | ||
| // Composio wraps the Stripe list envelope under its own `data` key. | ||
| serde_json::json!({ | ||
| "successful": true, | ||
| "data": { | ||
| "object": "list", | ||
| "has_more": true, | ||
| "data": [ | ||
| {"id": "ch_1", "object": "charge", "created": 1700000001, "amount": 500, "description": "Pro plan"}, | ||
| {"id": "ch_2", "object": "charge", "created": 1700000002, "amount": 900} | ||
| ] | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| #[test] | ||
| fn toolkit_and_action_match_composio_slug() { | ||
| let pipeline = pipeline(); | ||
| assert_eq!(pipeline.toolkit(), "stripe"); | ||
| assert_eq!(pipeline.action(), "STRIPE_LIST_ALL_CHARGES"); | ||
| } | ||
|
|
||
| #[test] | ||
| fn extract_page_reads_charges_and_cursor() { | ||
| let pipeline = pipeline(); | ||
| let page = pipeline.extract_page(&sample_payload(), None); | ||
| assert_eq!(page.items.len(), 2); | ||
| // `starting_after` for the next request is the last charge's id. | ||
| assert_eq!(page.next.as_deref(), Some("ch_2")); | ||
| } | ||
|
|
||
| #[test] | ||
| fn extract_page_stops_without_has_more() { | ||
| let pipeline = pipeline(); | ||
| let data = serde_json::json!({ | ||
| "data": {"has_more": false, "data": [{"id": "ch_9", "created": 1700000009}]} | ||
| }); | ||
| let page = pipeline.extract_page(&data, None); | ||
| assert_eq!(page.items.len(), 1); | ||
| assert!(page.next.is_none()); | ||
| } | ||
|
|
||
| #[test] | ||
| fn dedup_key_combines_id_and_created() { | ||
| let pipeline = pipeline(); | ||
| let item = serde_json::json!({"id": "ch_1", "created": 1700000001}); | ||
| assert_eq!( | ||
| pipeline.dedup_key(&item).as_deref(), | ||
| Some("ch_1@1700000001") | ||
| ); | ||
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn document_uses_stable_document_id_and_taint() { | ||
| let pipeline = pipeline(); | ||
| let mut state = SyncState::new("stripe", "conn-test"); | ||
| let raw = | ||
| serde_json::json!({"id": "ch_1", "created": 1700000001, "description": "Pro plan"}); | ||
| let item = SyncItem { | ||
| dedup_key: "ch_1@1700000001".into(), | ||
| sort_cursor: Some("1700000001".into()), | ||
| raw, | ||
| }; | ||
| let doc = pipeline | ||
| .document( | ||
| &SyncScope::flat(), | ||
| "conn-test", | ||
| item, | ||
| &pipeline.client, | ||
| &mut state, | ||
| ) | ||
| .await | ||
| .unwrap(); | ||
| // Stable upsert key: derived from the object id, not the per-run cursor. | ||
| assert_eq!(doc.document_id, "stripe:ch_1"); | ||
| assert_eq!(doc.namespace_skill_id, "stripe"); | ||
| assert_eq!(doc.toolkit, "stripe"); | ||
| assert_eq!(doc.title, "Pro plan"); | ||
| assert_eq!(doc.metadata["taint"], "external_sync"); | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Move tests to a sibling stripe_tests.rs file.
The #[cfg(test)] mod tests block is embedded directly in stripe.rs. As per coding guidelines, src/**/*.rs should "Keep tests in per-file <name>_tests.rs siblings, such as store.rs and store_tests.rs, rather than mixing tests into implementation files."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memory/sync/composio/providers/stripe.rs` around lines 169 - 265, The
tests are embedded in the implementation file instead of a sibling test module.
Move the entire #[cfg(test)] mod tests block, including pipeline,
sample_payload, and all test functions, into a sibling stripe_tests.rs file;
import the implementation symbols there as needed and remove the embedded test
module from stripe.rs.
Source: Coding guidelines
1 participant
Summary
Adds
StripeSyncPipeline, the tinycortex-side memory-sync pipeline for the Composiostripetoolkit. Previously connecting Stripe reportedACTIVEbut failed at sync withtinycortex sync does not support toolkit 'stripe'; this supplies the missing pipeline body so the slug becomes syncable.What it does
SyncPipelineandIncrementalSource, modeled on the document-shapedLinearSyncPipeline(charges are records with stable ids, not message threads).STRIPE_LIST_ALL_CHARGESusing Stripe's cursor pagination:limit+starting_after=<last charge id>, continuing only while the list envelope reportshas_more.document()helper, so the upsert key is the stable Stripe object id —document_id = stripe:<charge_id>withmetadata.taint = "external_sync". Thedocument_idis never derived from a per-run cursor, avoiding the duplicate-charge bug class fixed by fix(memory): use stable document_id as sync upsert key (fixes #4947 Bug 2 secret-guard sync failure) openhuman#4953. Thededup_keycombinesid@createdfor freshness; the upsert key stays the bare id.providers/mod.rs,composio/mod.rs, andsync/mod.rsexactly like the six existing pipelines.Reference followed
src/memory/sync/composio/providers/linear.rs(closest document-shaped pipeline) plus the sharedproviders/common.rsdocument()/first_array()/pick_str()helpers.Tests run
stripe.rs:toolkit()/action()slug,extract_pageon a sample Composio-wrapped Stripe list payload (charge extraction +starting_aftercursor +has_morestop), stablededup_key, and a stabledocument_idfromdocument().cargo fmt --all -- --check— clean.cargo test --features sync sync::composio— 22 passed (incl. 5 new Stripe tests).cargo test --features sync --test composio_sync_mock— 12 passed (no regressions).Part of #78 (tinycortex pipeline body; openhuman wiring is the follow-up step)
🤖 Generated with Claude Code
Summary by CodeRabbit