Introduction​

Generating long-form audio content—audiobooks spanning hours, educational courses, or extended podcasts—presents unique challenges: API rate limits, network failures, resource constraints, and the sheer duration of processing. This article explores a production-ready architecture for long-running TTS pipelines that can gracefully handle long-form generation tasks, resume after failures, and maintain state across distributed systems.

Built with LangGraph, the system orchestrates complex workflows involving AI content generation (DeepSeek), text-to-speech conversion (OpenAI TTS), and distributed storage (Cloudflare R2). The key innovation: PostgreSQL checkpointing enables resumable execution, making it possible to generate 5-30+ minute audio segments reliably, even when individual API calls or processing steps fail.

The Challenge: Long-Form Audio at Scale​

Why Long-Running Pipelines Are Hard​

Traditional TTS approaches fail at scale:

1. Time Constraints: A 30-minute audio narrative requires ~4,500 words, chunked into 10-15 API calls, taking 2-5 minutes to generate
2. Failure Points: Each step (text generation, chunking, TTS, storage) can fail independently
3. Memory Pressure: Holding all audio segments in memory for hours is impractical
4. Cost Management: Retrying from scratch wastes API credits and compute time
5. State Loss: Without persistence, crashes mean starting over

Our Solution: Stateful Orchestration

• LangGraph manages workflow state transitions
• PostgreSQL persists checkpoints after each successful step
• R2 provides durable storage for completed segments
• Resumable execution using thread_id for job recovery

System Overview​

The pipeline orchestrates three main workflows:

1. Research Generation: Structured content research using DeepSeek
2. Narrative Text Generation: Long-form content creation with context awareness
3. Audio Synthesis: Text-to-speech conversion with OpenAI TTS and Cloudflare R2 storage

Tech Stack​

• LangGraph: State machine orchestration with built-in checkpointing
• DeepSeek: Long-form text generation (deepseek-chat, 2500+ token outputs)
• OpenAI TTS: Streaming audio synthesis (gpt-4o-mini-tts, 4096 char limit per request)
• PostgreSQL: Durable checkpointing for long-running jobs (Neon serverless for production)
• Cloudflare R2: S3-compatible storage with zero egress fees (critical for multi-GB audio)
• FastAPI: Async REST API for non-blocking long operations
• Docker: Containerized deployment with ffmpeg for audio merging

Why This Stack for Long-Running Jobs:

• Postgres checkpointing: Resume from any point in the workflow (text generation → chunking → TTS → upload)
• Streaming TTS: Memory-efficient direct-to-disk writes (no buffering entire audio in RAM)
• R2 durability: Segments uploaded immediately, survive process crashes
• Async execution: Non-blocking background processing for hours-long jobs

Architecture Patterns​

1. Core LangGraph State Machine​

The system implements three distinct LangGraph workflows, each optimized for specific tasks.

2. Research Generation Pipeline​

The research pipeline generates structured research content using a focused LangGraph workflow.

Key Features:

• Low temperature (0.3) for factual accuracy
• Structured JSON output with validation
• Evidence level classification (A/B/C)
• Relevance scoring for topic matching

3. Long-Form Text Generation Pipeline​

The most sophisticated workflow, supporting both full generation and audio-only modes.

Conditional Routing Logic:

def should_skip_text_generation(state: TextState) -> str:
    """Route to text generation or skip to audio."""
    if state.get("existing_content") and state["existing_content"].get("text"):
        return "chunk_text"  # Audio-only mode
    return "generate_text"  # Full generation

def should_generate_audio(state: TextState) -> str:
    """Route to audio generation or end."""
    if state.get("generate_audio", True):
        return "chunk_text"
    return END  # Text-only mode

4. Audio Generation Pipeline (Standalone)​

A simplified pipeline for generic long-form narration.

Iterative Chunk Processing:

The system uses a recursive edge pattern for processing chunks:

g.add_conditional_edges(
    "tts_one_chunk",
    edge_should_continue,
    {
        "tts_one_chunk": "tts_one_chunk",  # Loop back
        "finalize": "finalize",             # Exit loop
    },
)

def edge_should_continue(state: JobState) -> str:
    if state["chunk_index"] < len(state["chunks"]):
        return "tts_one_chunk"
    return "finalize"

Deep Dive: Key Architectural Components​

State Management​

LangGraph uses typed state dictionaries for type safety and IDE support:

class TextState(TypedDict):
    # Input metadata
    content_id: int
    title: str
    content_type: str
    language: str
    target_duration_minutes: int | None
    
    # Generation data
    research_items: list[dict]
    existing_content: dict | None
    generated_text: str | None
    
    # TTS fields
    voice: str
    chunks: List[str]
    segment_urls: List[str]
    manifest_url: Optional[str]
    audio_url: Optional[str]
    
    # Control flow
    generate_audio: bool
    database_saved: bool
    error: str | None

Postgres Checkpointing: The Key to Long-Running Resilience​

For long-running jobs, checkpointing is non-negotiable. Without it, a network glitch at minute 25 of a 30-minute generation means restarting from scratch.

How Checkpointing Works:

async def run_pipeline(state: TextState, thread_id: str):
    db_url = os.getenv("DATABASE_URL")
    
    async with AsyncPostgresSaver.from_conn_string(db_url) as checkpointer:
        await checkpointer.setup()  # Creates checkpoint tables
        app = build_graph(checkpointer=checkpointer)
        config = {"configurable": {"thread_id": thread_id}}
        
        # LangGraph automatically saves state after each node execution
        final_state = await app.ainvoke(state, config=config)
        return final_state

What Gets Checkpointed:

• Complete state dictionary after each node
• Edge transitions and routing decisions
• Timestamps and execution metadata
• Partial results (generated text, uploaded segment URLs)

Recovery Example:

# Job crashes after generating 8 of 12 TTS segments
# Resume with same thread_id:
final_state = await run_pipeline(initial_state, thread_id="job-12345")

# LangGraph:
# 1. Loads last checkpoint from Postgres
# 2. Sees 8 segments already uploaded to R2
# 3. Continues from segment 9
# 4. Completes remaining 4 segments

Production Benefits:

• Cost Savings: No wasted API calls on retry
• Time Efficiency: Resume from 80% complete, not 0%
• Reliability: Transient failures (rate limits, timeouts) don't kill long-form jobs
• Observability: Query checkpoint table to monitor progress
• Parallel Execution: Multiple jobs with different thread_id values

Text Chunking Algorithm: Optimizing for Long-Form Narration​

For 30-minute audio (4,500+ words), naive chunking creates jarring transitions. Our algorithm balances API constraints with narrative flow:

Constraints:

• OpenAI TTS: 4,096 character limit per request
• Target: ~4,000 chars per chunk (safety margin)
• Goal: Natural pauses at paragraph/sentence boundaries

Strategy:

def chunk_text(text: str, max_chars: int = 4000) -> List[str]:
    """
    Multi-level chunking for long-form content:
    
    1. Split by paragraphs (\n\n) - natural topic boundaries
    2. Accumulate paragraphs until approaching 4K limit
    3. If single paragraph > 4K, split by sentences
    4. If single sentence > 4K, split mid-sentence (rare edge case)
    
    Result: 10-15 chunks for 30-min audio, each ending at natural pause
    """
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks = []
    buf = []
    
    for p in paragraphs:
        candidate = "\n\n".join(buf + [p]) if buf else p
        if len(candidate) <= max_chars:
            buf.append(p)
        else:
            if not buf:
                # Paragraph too large - split by sentences
                sentences = re.split(r"(?<=[.!?])\s+", p)
                # Accumulate sentences with same logic...
            else:
                chunks.append("\n\n".join(buf))
                buf = [p]
    
    return chunks

Why This Matters for Long-Form:

• Seamless Merging: Chunk boundaries at natural pauses prevent audio glitches
• Even Distribution: Avoids tiny final chunks (better for progress tracking)
• Memory Efficiency: Process one chunk at a time, not entire 4,500-word text
• Resumability: Each chunk is independent; can resume mid-sequence

OpenAI TTS Streaming​

Efficient audio generation using streaming responses:

async def node_tts_one_chunk(state: JobState) -> JobState:
    chunk_text = state["chunks"][state["chunk_index"]]
    segment_path = f"segment_{state['chunk_index']:04d}.mp3"
    
    client = OpenAI()
    
    # Stream directly to disk (memory efficient)
    with client.audio.speech.with_streaming_response.create(
        model="gpt-4o-mini-tts",
        voice=state["voice"],
        input=chunk_text,
        response_format="mp3",
    ) as response:
        response.stream_to_file(segment_path)
    
    # Upload to R2
    r2_url = upload_to_r2(segment_path, state["job_id"])
    
    return {
        **state,
        "segment_urls": [*state["segment_urls"], r2_url],
        "chunk_index": state["chunk_index"] + 1,
    }

Audio Merging Strategy​

The system uses ffmpeg for high-quality concatenation:

async def node_generate_audio(state: TextState) -> TextState:
    # Generate all segments...
    
    # Create concat list for ffmpeg
    file_list_path.write_text(
        "\n".join(f"file '{segment}'" for segment in segment_paths)
    )
    
    # Merge using ffmpeg (codec copy - fast and lossless)
    subprocess.run([
        "ffmpeg", "-f", "concat", "-safe", "0",
        "-i", str(file_list_path),
        "-c", "copy",  # No re-encoding
        str(merged_path)
    ])
    
    # Fallback to binary concatenation if ffmpeg unavailable
    if not merged_path.exists():
        with open(merged_path, "wb") as merged:
            for segment in segment_paths:
                merged.write(segment.read_bytes())

Cloudflare R2 Integration​

S3-compatible storage for globally distributed audio:

def get_r2_client():
    return boto3.client(
        's3',
        endpoint_url=f'https://{R2_ACCOUNT_ID}.r2.cloudflarestorage.com',
        aws_access_key_id=R2_ACCESS_KEY_ID,
        aws_secret_access_key=R2_SECRET_ACCESS_KEY,
        config=Config(signature_version='s3v4'),
    )

def upload_to_r2(file_path: Path, job_id: str) -> str:
    key = f"{job_id}/{file_path.name}"
    
    client.put_object(
        Bucket=R2_BUCKET_NAME,
        Key=key,
        Body=file_path.read_bytes(),
        ContentType='audio/mpeg',
    )
    
    return f"{R2_PUBLIC_DOMAIN}/{key}"

Structured Content Generation​

Narrative Architecture Framework​

The system implements a flexible content framework with customizable sections:

Key Components:

1. Introduction (2-3 min): Hook the listener and set expectations
2. Context: Background information and relevance
3. Core Content: Main topic introduction with clear structure
4. Examples: Concrete illustrations and case studies
5. Deep Dive: Detailed exploration of key concepts
6. Applications: Practical use cases and implementation
7. Advanced Topics: Nuanced discussion for engaged learners
8. Synthesis: Connect all concepts together
9. Takeaways: Summary of key points
10. Conclusion: Clear closing and next steps

Dynamic Content Adaptation​

def build_content_prompt(state: TextState) -> str:
    minutes = state.get("target_duration_minutes") or 5
    target_words = int(minutes * 150)  # 150 words per minute narration
    
    content_type = state.get("content_type")
    
    # Select architecture based on content type
    architecture = generate_content_architecture(content_type)
    
    return f"""
Create a {state['language']} narrative for audio:

TOPIC: {state['title']}
TYPE: {content_type}
TARGET: {target_words} words ({minutes} minutes)

{architecture}

RESEARCH CONTEXT:
{format_research_summary(state['research_items'])}

Requirements:
- Plain text only (no markdown)
- Natural paragraph breaks
- Engaging, clear tone
- Appropriate language for audio listening
"""

API Endpoints​

FastAPI Service Layer​

Endpoint Implementations:

@app.post("/api/research/generate")
async def research_endpoint(req: ResearchRequest):
    """Generate research context using LangGraph + DeepSeek."""
    return await generate_research(req)

@app.post("/api/text/generate")
async def text_endpoint(req: TextGenerationRequest):
    """Generate long-form text content (text-only mode)."""
    return await generate_text(req)

@app.post("/api/audio/generate")
async def audio_endpoint(req: TextGenerationRequest):
    """Generate audio from existing content (audio-only mode)."""
    return await generate_audio(req)

@app.post("/api/tts/generate")
async def tts_endpoint(req: TTSRequest, background_tasks: BackgroundTasks):
    """Generic TTS generation (fire-and-forget)."""
    return await generate_tts(req, background_tasks)

Deployment Architecture​

Docker Containerization​

FROM python:3.12-slim

WORKDIR /app

# Install ffmpeg for audio merging
RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Expose port
EXPOSE 8080

# Run FastAPI server
CMD ["uvicorn", "langgraph_server:app", "--host", "0.0.0.0", "--port", "8080"]

Environment Configuration​

# AI Services
DEEPSEEK_API_KEY=sk-...
OPENAI_API_KEY=sk-...

# Database (Neon Postgres)
DATABASE_URL=postgresql://user:pass@host/db?sslmode=require

# Cloudflare R2
R2_ACCOUNT_ID=...
R2_ACCESS_KEY_ID=...
R2_SECRET_ACCESS_KEY=...
R2_BUCKET_NAME=longform-tts
R2_PUBLIC_DOMAIN=https://pub-longform-tts.r2.dev

Cloudflare Workers Deployment​

# wrangler.toml
name = "langgraph-tts"
compatibility_date = "2024-01-01"

[build]
command = "docker build -t langgraph-tts ."

[[services]]
name = "langgraph-tts"
image = "langgraph-tts:latest"

[env]
PORT = "8080"

[[r2_buckets]]
binding = "LONGFORM_TTS"
bucket_name = "longform-tts"

Production Considerations​

Performance Metrics for Long-Running Jobs​

Benchmarks (30-minute audio generation):

Long-Running Job Profile:

# Example: 2-hour audiobook chapter
text_length = 18,000 words
chunks = 45  # ~4,000 chars each
tts_time = 45 * 15s = 11.25 minutes
text_gen_time = 2-3 minutes
total_time = ~15 minutes for 2-hour audio

# Checkpoint frequency:
# - 1 after text generation
# - 45 after each TTS segment
# - 1 after merge
# Total: 47 recovery points

Failure Recovery Times:

• Crash at 80% complete → Resume in 1–2 seconds, continue from segment 36/45
• Network timeout on segment 20 → Retry only segment 20, not segments 1–19
• Database connection loss → Reconnect and load last checkpoint (<500ms)

Error Handling & Resilience​

async def node_generate_text(state: TextState) -> TextState:
    try:
        llm = ChatDeepSeek(model="deepseek-chat", temperature=0.7, max_tokens=2500)
        prompt = build_therapeutic_prompt(state)
        
        resp = await llm.ainvoke([HumanMessage(content=prompt)])
        text = clean_for_tts(resp.content)
        
        return {**state, "generated_text": text, "error": None}
    except Exception as e:
        print(f"❌ Text generation failed: {e}")
        return {**state, "error": str(e)}

Monitoring & Observability​

Key metrics to track:

1. Generation Metrics:

   • Text generation latency (DeepSeek)
   • TTS latency per chunk (OpenAI)
   • Total pipeline duration

2. Quality Metrics:

   • Text length vs target duration
   • Chunk count and size distribution
   • Audio segment file sizes

3. Infrastructure Metrics:

   • R2 upload success rate
   • Database checkpoint writes
   • ffmpeg merge success rate

4. Cost Metrics:

   • DeepSeek token usage
   • OpenAI TTS character count
   • R2 storage and bandwidth

Scaling Patterns​

Horizontal Scaling:

• FastAPI instances behind load balancer
• Stateless design (state in Postgres)
• R2 for distributed storage

Batch Processing:

async def batch_generate_audio(goal_ids: List[int]):
    """Process multiple goals in parallel."""
    tasks = [run_pipeline(build_state(id), f"batch-{id}") for id in goal_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

Queue-Based Processing:

• Use background tasks for long-running jobs
• Celery/Redis for distributed task queue
• Webhook callbacks for completion notifications

Performance Optimization​

Chunking Optimization​

# Optimize chunk size for TTS quality vs API limits
OPTIMAL_CHUNK_SIZE = 3000  # Sweet spot for natural pauses

# Parallel TTS generation (with rate limiting)
async def parallel_tts_generation(chunks: List[str], max_concurrent: int = 3):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def generate_with_limit(chunk, index):
        async with semaphore:
            return await generate_tts_segment(chunk, index)
    
    tasks = [generate_with_limit(c, i) for i, c in enumerate(chunks)]
    return await asyncio.gather(*tasks)

Caching Strategy​

# Cache research results for similar goals
@lru_cache(maxsize=100)
def get_research_for_goal_type(therapeutic_type: str, age: int):
    """Cache research by type + age bracket."""
    return fetch_research(therapeutic_type, age)

# Cache text generation for re-use
async def get_or_generate_text(goal_id: int):
    existing = await db.fetch_story(goal_id)
    if existing and existing.created_at > datetime.now() - timedelta(days=7):
        return existing.text
    return await generate_new_text(goal_id)

Testing Strategy​

Unit Tests​

def test_chunk_text_respects_limit():
    long_text = "word " * 2000
    chunks = chunk_text(long_text, max_chars=4000)
    
    for chunk in chunks:
        assert len(chunk) <= 4000

def test_clean_for_tts_removes_markdown():
    text = "# Title\n\n**bold** and `code`"
    cleaned = clean_for_tts(text)
    assert "#" not in cleaned
    assert "**" not in cleaned
    assert "`" not in cleaned

Integration Tests​

@pytest.mark.asyncio
async def test_full_pipeline():
    state = {
        "goal_id": 1,
        "goal_title": "Test anxiety reduction",
        "therapeutic_goal_type": "anxiety_reduction",
        "age": 8,
        # ... other fields
    }
    
    result = await run_pipeline(state, "test-thread-1")
    
    assert result["generated_text"] is not None
    assert len(result["chunks"]) > 0
    assert result["audio_url"] is not None
    assert result["error"] is None

Lessons Learned​

1. State Design is Critical​

• Use TypedDict for type safety
• Keep state flat (avoid deep nesting)
• Include metadata for debugging (timestamps, IDs)

2. Checkpoint Strategically​

• Not all workflows need checkpointing
• Audio-only mode: disable checkpoints to avoid schema issues
• Use thread_id conventions: {workflow}-{entity_id}-{timestamp}

3. Error Recovery​

• Graceful degradation (segments work even if merge fails)
• Fallback strategies (binary concat if ffmpeg unavailable)
• Preserve partial results (individual segments in R2)

4. Cost Management​

• Monitor token usage (DeepSeek is cost-effective at $0.14/$0.28 per 1M tokens)
• OpenAI TTS: $15 per 1M characters
• R2 storage: $0.015/GB/month (much cheaper than S3)

5. Content Quality​

• Structured frameworks improve consistency
• Repetition aids retention and comprehension
• Audience-appropriate language is crucial for engagement

Future Enhancements​

1. Multi-Voice Narratives​

# Support character dialogue with different voices
voices = {
    "narrator": "cedar",
    "child_character": "nova",
    "parent_character": "marin"
}

2. Emotion-Adaptive TTS​

# Adjust voice parameters based on content emotion
def get_tts_params(text: str) -> dict:
    sentiment = analyze_sentiment(text)
    
    if sentiment == "calm":
        return {"speed": 0.9, "pitch": 0}
    elif sentiment == "energetic":
        return {"speed": 1.1, "pitch": 2}

3. Real-Time Streaming​

# Stream audio as it's generated (SSE)
async def stream_audio_generation(goal_id: int):
    async for chunk_url in generate_audio_stream(goal_id):
        yield f"data: {json.dumps({'chunk_url': chunk_url})}\n\n"

4. Multilingual Support​

• Expand beyond Romanian and English
• Voice selection per language
• Cultural adaptation of content frameworks

Conclusion​

This LangGraph-based TTS architecture demonstrates several key patterns:

1. Composable Workflows: Three distinct pipelines sharing common components
2. Conditional Routing: Smart flow control based on state
3. Durable Execution: PostgreSQL checkpointing for resilience
4. Streaming Efficiency: Direct-to-disk TTS for memory optimization
5. Distributed Storage: R2 for globally accessible audio

The system successfully processes 5-30+ minute long-form narratives (up to 7,000+ words), generating research-backed content, converting to high-quality audio, and delivering via CDN—all while maintaining resumability after failures and full observability.

Real-World Performance:

• 30-minute generation: 12-15 TTS chunks, ~3-5 minutes total processing time
• Failure recovery: Resume from any checkpoint in <1 second
• Cost efficiency: $0.02-$0.07 per 30-minute audio (DeepSeek + OpenAI TTS)
• Throughput: 10+ concurrent jobs on single instance

Key Takeaways for Long-Running Pipelines:

• LangGraph + Postgres checkpointing is essential for long-form workflows
• Streaming TTS to disk prevents memory exhaustion on long generations
• Smart chunking (4K chars) balances API limits with narrative coherence
• Immediate R2 uploads ensure partial results survive crashes
• Async architecture enables fire-and-forget long operations
• Thread-based recovery makes interrupted jobs trivial to resume

The architecture scales to long-form audio generation: audiobooks (10+ hours), comprehensive courses, documentary narration, or serialized storytelling—any use case where reliability and resumability are non-negotiable.

References​

• LangGraph Documentation
• DeepSeek API
• OpenAI TTS API
• Cloudflare R2
• PostgreSQL Async Python

This architecture powers long-form audio generation, combining LangGraph orchestration, OpenAI TTS streaming, and distributed storage for production-ready AI audio systems.

TL;DR​

This guide demonstrates real-world implementation of DeFiTuna limit orders on Solana mainnet, focusing on:

• Direct RPC Interaction: Building and submitting transactions without high-level SDKs
• On-Chain Limit Order Storage: How limit order parameters are encoded in account data
• Position Lifecycle: Open position → Set limits → Close position with actual mainnet transactions
• Account Derivation: PDA calculations for spot positions and associated token accounts
• DeFiTuna SDK Patterns: Insights from the official TypeScript/Rust SDK implementation

Live Mainnet Transactions:

• Position Created: 4J9XJxHUDmSUV...
• Limit Orders Set: 5ouHm9mVroRGz...
• Position Closed: 44aWvGvPEWGK...

Introduction​

DeFiTuna combines Orca Whirlpools with on-chain limit orders, enabling automated trading triggers without requiring active monitoring. Unlike traditional AMMs where you passively provide liquidity, DeFiTuna positions execute predefined trades when price conditions are met.

This article walks through the complete implementation—from deriving PDAs to encoding instruction data—using Rust with solana-sdk and insights from the DeFiTuna SDK.

Architecture Overview​

DeFiTuna + Orca Integration​

Key Insight: DeFiTuna is a Protocol Layer​

DeFiTuna doesn't implement its own AMM. Instead, it:

1. Wraps existing AMMs (Orca Whirlpools, Fusion pools)
2. Adds limit order logic on top
3. Manages leveraged positions and lending vaults

On-Chain Account Structure​

Position Account Data Layout (346 bytes)​

From our mainnet position 8wvKhHXHfzY4eQZTyK4kTfUtGj46XX5UX8P4S5kBbJ5:

Offset   Field                    Type     Bytes   Description
------   -----                    ----     -----   -----------
0-8      Discriminator            u64      8       Account type identifier
8-40     Authority                Pubkey   32      Position owner
40-72    Pool                     Pubkey   32      Orca Whirlpool address
72-73    Position Token           u8       1       0=TokenA, 1=TokenB
73-74    Collateral Token         u8       1       0=TokenA, 1=TokenB
92-100   Amount                   u64      8       Position size
100-108  Borrowed                 u64      8       Borrowed amount
184-200  Lower Limit √Price       u128     16      Buy trigger price
200-216  Upper Limit √Price       u128     16      Sell trigger price

Hex dump from mainnet:

00b8:   60 e4 c0 d6 1c 8e 68 01  00 00 00 00 00 00 00 00   Lower limit
00c8:   70 17 34 50 e5 c9 7c 01  00 00 00 00 00 00 00 00   Upper limit

These bytes encode:

• Lower: 6651068162312125808640 → $130
• Upper: 7024310870365581606912 → $145

Implementation: Rust Binaries​

Project Structure​

bots/defituna-bot/
├── Cargo.toml
├── .env                    # Configuration
├── defituna.json          # Program IDL
└── src/
    ├── config.rs          # Shared config
    └── bin/
        ├── open_spot_position.rs    # Create position
        ├── set_limit_orders.rs      # Configure triggers
        ├── check_position.rs        # Query state
        └── close_position.rs        # Cleanup

1. Opening a Spot Position​

File: src/bin/open_spot_position.rs

use solana_sdk::{
    instruction::{AccountMeta, Instruction},
    signature::{Keypair, Signer},
    transaction::Transaction,
};
use spl_associated_token_account::get_associated_token_address;

// DeFiTuna program ID (mainnet)
const DEFITUNA_PROGRAM: &str = "tuna4uSQZncNeeiAMKbstuxA9CUkHH6HmC64wgmnogD";

// Orca SOL/USDC Whirlpool
const WHIRLPOOL: &str = "Czfq3xZZDmsdGdUyrNLtRhGc47cXcZtLG4crryfu44zE";

fn main() -> Result<()> {
    let program_id = Pubkey::from_str(DEFITUNA_PROGRAM)?;
    let whirlpool = Pubkey::from_str(WHIRLPOOL)?;
    let authority = executor_keypair.pubkey();
    
    // Derive spot position PDA
    let (tuna_spot_position, _bump) = Pubkey::find_program_address(
        &[
            b"tuna_spot_position",
            authority.as_ref(),
            whirlpool.as_ref(),
        ],
        &program_id,
    );
    
    // Token program IDs
    let token_program_a = spl_token::ID; // SOL uses standard program
    let token_program_b = spl_token::ID; // USDC uses standard program
    
    // Derive associated token accounts for position
    let tuna_position_ata_a = get_associated_token_address(
        &tuna_spot_position,
        &SOL_MINT
    );
    let tuna_position_ata_b = get_associated_token_address(
        &tuna_spot_position,
        &USDC_MINT
    );
    
    // Build instruction
    // Discriminator from IDL: [87, 208, 173, 48, 231, 62, 210, 220]
    let mut data = vec![87, 208, 173, 48, 231, 62, 210, 220];
    
    // Args: position_token (PoolToken::A=0), collateral_token (PoolToken::B=1)
    data.push(0); // Trading SOL (position_token = A)
    data.push(1); // Collateral is USDC (collateral_token = B)
    
    let instruction = Instruction {
        program_id,
        accounts: vec![
            AccountMeta::new(authority, true),                  // authority (signer, writable)
            AccountMeta::new_readonly(SOL_MINT, false),        // mint_a
            AccountMeta::new_readonly(USDC_MINT, false),       // mint_b
            AccountMeta::new_readonly(token_program_a, false), // token_program_a
            AccountMeta::new_readonly(token_program_b, false), // token_program_b
            AccountMeta::new(tuna_spot_position, false),       // tuna_position (writable)
            AccountMeta::new(tuna_position_ata_a, false),      // tuna_position_ata_a
            AccountMeta::new(tuna_position_ata_b, false),      // tuna_position_ata_b
            AccountMeta::new_readonly(whirlpool, false),       // pool (Orca Whirlpool)
            AccountMeta::new_readonly(system_program::ID, false), // system_program
            AccountMeta::new_readonly(
                spl_associated_token_account::ID, 
                false
            ), // associated_token_program
        ],
        data,
    };
    
    // Create and sign transaction
    let recent_blockhash = rpc_client.get_latest_blockhash()?;
    let transaction = Transaction::new_signed_with_payer(
        &[instruction],
        Some(&authority),
        &[&executor_keypair],
        recent_blockhash,
    );
    
    // Send to RPC
    let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
    println!("Position created: {}", signature);
    
    Ok(())
}

Key Points:

• No collateral required to open position (just creates account structure)
• Position PDA seeds: ["tuna_spot_position", authority, whirlpool]
• Instruction creates position account + 2 associated token accounts
• Cost: ~0.00329904 SOL rent + ~0.000005 SOL gas

2. Setting Limit Orders​

File: src/bin/set_limit_orders.rs

/// Convert price to sqrt_price format
/// Formula: sqrt_price = sqrt(price) * 2^64, adjusted for decimals
fn price_to_sqrt_price(price: f64, decimals_a: u8, decimals_b: u8) -> u128 {
    let decimal_diff = decimals_a as i32 - decimals_b as i32;
    let adjusted_price = if decimal_diff >= 0 {
        price * 10_f64.powi(decimal_diff)
    } else {
        price / 10_f64.powi(-decimal_diff)
    };
    
    let sqrt_price_f64 = adjusted_price.sqrt() * (1u128 << 64) as f64;
    sqrt_price_f64 as u128
}

fn main() -> Result<()> {
    // Derive same position PDA
    let (tuna_spot_position, _) = Pubkey::find_program_address(
        &[b"tuna_spot_position", authority.as_ref(), whirlpool.as_ref()],
        &program_id,
    );
    
    // Set limit prices
    let lower_price = 130.0; // Buy if SOL drops to $130
    let upper_price = 145.0; // Sell if SOL rises to $145
    
    // Convert to sqrt_price (SOL=9 decimals, USDC=6 decimals)
    let lower_sqrt_price = price_to_sqrt_price(lower_price, 9, 6);
    let upper_sqrt_price = price_to_sqrt_price(upper_price, 9, 6);
    
    // Build instruction
    // Discriminator: [10, 180, 19, 205, 169, 133, 52, 118]
    let mut data = vec![10, 180, 19, 205, 169, 133, 52, 118];
    
    // Args: lower_limit_order_sqrt_price (u128), upper_limit_order_sqrt_price (u128)
    data.extend_from_slice(&lower_sqrt_price.to_le_bytes());
    data.extend_from_slice(&upper_sqrt_price.to_le_bytes());
    
    let instruction = Instruction {
        program_id,
        accounts: vec![
            AccountMeta::new_readonly(authority, true),    // authority (signer)
            AccountMeta::new(tuna_spot_position, false),  // tuna_position (writable)
        ],
        data,
    };
    
    let transaction = Transaction::new_signed_with_payer(
        &[instruction],
        Some(&authority),
        &[&executor_keypair],
        recent_blockhash,
    );
    
    let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
    println!("Limit orders set: {}", signature);
    
    Ok(())
}

Output (mainnet):

Lower (buy): $130 → sqrt_price: 6651068162312125808640
Upper (sell): $145 → sqrt_price: 7024310870365581606912

Verification on-chain:

solana account 8wvKhHXHfzY4eQZTyK4kTfUtGj46XX5UX8P4S5kBbJ5

# Bytes 184-200 (lower limit):
60 e4 c0 d6 1c 8e 68 01 00 00 00 00 00 00 00 00

# Bytes 200-216 (upper limit):
70 17 34 50 e5 c9 7c 01 00 00 00 00 00 00 00 00

3. Closing a Position​

File: src/bin/close_position.rs

fn main() -> Result<()> {
    // Derive position PDA (same as open)
    let (tuna_spot_position, _) = Pubkey::find_program_address(
        &[b"tuna_spot_position", authority.as_ref(), whirlpool.as_ref()],
        &program_id,
    );
    
    // Build close instruction
    // Discriminator: [4, 189, 171, 84, 110, 220, 10, 8]
    let data = vec![4, 189, 171, 84, 110, 220, 10, 8];
    
    let instruction = Instruction {
        program_id,
        accounts: vec![
            AccountMeta::new(authority, true),                 // authority
            AccountMeta::new_readonly(SOL_MINT, false),       // mint_a
            AccountMeta::new_readonly(USDC_MINT, false),      // mint_b
            AccountMeta::new_readonly(spl_token::ID, false),  // token_program_a
            AccountMeta::new_readonly(spl_token::ID, false),  // token_program_b
            AccountMeta::new(tuna_spot_position, false),      // tuna_position
            AccountMeta::new(tuna_position_ata_a, false),     // tuna_position_ata_a
            AccountMeta::new(tuna_position_ata_b, false),     // tuna_position_ata_b
        ],
        data,
    };
    
    let signature = rpc_client.send_and_confirm_transaction(&transaction)?;
    println!("Position closed, rent recovered: {}", signature);
    
    Ok(())
}

Result: Rent (0.00329904 SOL) returned to wallet, position account deleted.

DeFiTuna SDK Patterns​

TypeScript SDK Structure​

From DefiTuna/tuna-sdk repository:

// ts-sdk/client/src/txbuilder/openTunaSpotPosition.ts
export async function openTunaSpotPositionInstructions(
  rpc: Rpc<GetAccountInfoApi & GetMultipleAccountsApi>,
  authority: TransactionSigner,
  poolAddress: Address,
  args: OpenTunaSpotPositionInstructionDataArgs,
): Promise<IInstruction[]> {
  // 1. Derive position PDA
  const tunaPositionAddress = (
    await getTunaSpotPositionAddress(authority.address, poolAddress)
  )[0];

  // 2. Get associated token accounts
  const tunaPositionAtaA = (
    await findAssociatedTokenPda({
      owner: tunaPositionAddress,
      mint: mintA.address,
      tokenProgram: mintA.programAddress,
    })
  )[0];

  const tunaPositionAtaB = (
    await findAssociatedTokenPda({
      owner: tunaPositionAddress,
      mint: mintB.address,
      tokenProgram: mintB.programAddress,
    })
  )[0];

  // 3. Build instruction
  return getOpenTunaSpotPositionInstruction({
    authority,
    mintA: mintA.address,
    mintB: mintB.address,
    tokenProgramA: mintA.programAddress,
    tokenProgramB: mintB.programAddress,
    tunaPosition: tunaPositionAddress,
    tunaPositionAtaA,
    tunaPositionAtaB,
    pool: poolAddress,
    associatedTokenProgram: ASSOCIATED_TOKEN_PROGRAM_ADDRESS,
    ...args,
  });
}

Rust SDK Patterns​

From rust-sdk/client/src/txbuilder/open_tuna_spot_position.rs:

pub fn open_tuna_spot_position_instructions(
    rpc: &RpcClient,
    authority: &Pubkey,
    pool_address: &Pubkey,
    args: OpenTunaSpotPositionInstructionArgs,
) -> Result<Vec<Instruction>> {
    // 1. Fetch pool account to determine token mints
    let pool_account = rpc.get_account(pool_address)?;
    
    // Decode based on program owner
    let (mint_a_address, mint_b_address) = if pool_account.owner == FUSIONAMM_ID {
        let pool: FusionPool = decode_account(&pool_account)?;
        (pool.token_mint_a, pool.token_mint_b)
    } else if pool_account.owner == WHIRLPOOL_ID {
        let pool: Whirlpool = decode_account(&pool_account)?;
        (pool.token_mint_a, pool.token_mint_b)
    } else {
        return Err(anyhow!("Unsupported pool type"));
    };

    // 2. Get mint accounts to determine token programs
    let mint_accounts = rpc.get_multiple_accounts(&[
        mint_a_address.into(),
        mint_b_address.into()
    ])?;
    
    let mint_a_account = mint_accounts[0].as_ref()
        .ok_or(anyhow!("Token A mint not found"))?;
    let mint_b_account = mint_accounts[1].as_ref()
        .ok_or(anyhow!("Token B mint not found"))?;

    // 3. Build instruction
    Ok(vec![open_tuna_spot_position_instruction(
        authority,
        pool_address,
        &mint_a_address,
        &mint_b_address,
        &mint_a_account.owner, // Token program
        &mint_b_account.owner,
        args,
    )])
}

Key SDK Features:

1. Pool Type Detection: Automatically handles Orca vs Fusion pools
2. Token Program Discovery: Supports both SPL Token and Token-2022
3. Account Pre-Validation: Checks accounts exist before building transaction
4. Error Handling: Detailed error messages for debugging

Solana RPC Interaction Patterns​

1. Account Fetching with Commitment​

use solana_client::rpc_client::RpcClient;
use solana_sdk::commitment_config::CommitmentConfig;

let rpc_client = RpcClient::new_with_commitment(
    "https://api.mainnet-beta.solana.com",
    CommitmentConfig::confirmed()
);

// Fetch account with specific commitment
let account = rpc_client.get_account_with_commitment(
    &position_address,
    CommitmentConfig::finalized()
)?;

if let Some(account) = account.value {
    println!("Account exists: {} bytes", account.data.len());
}

2. Transaction Simulation Before Sending​

// Build transaction
let transaction = Transaction::new_signed_with_payer(
    &[instruction],
    Some(&payer.pubkey()),
    &[&payer],
    recent_blockhash,
);

// Simulate first
match rpc_client.simulate_transaction(&transaction) {
    Ok(result) => {
        if let Some(err) = result.value.err {
            println!("Simulation failed: {:?}", err);
            if let Some(logs) = result.value.logs {
                for log in logs {
                    println!("  {}", log);
                }
            }
            return Err(anyhow!("Simulation error"));
        }
        println!("✅ Simulation successful");
    }
    Err(e) => {
        println!("RPC error during simulation: {}", e);
        return Err(e.into());
    }
}

// Now send for real
let signature = rpc_client.send_and_confirm_transaction(&transaction)?;

3. Parsing Account Data​

fn parse_position_limits(account_data: &[u8]) -> Result<(u128, u128)> {
    if account_data.len() < 216 {
        return Err(anyhow!("Account data too short"));
    }
    
    // Extract limit order sqrt prices
    let lower_bytes: [u8; 16] = account_data[184..200]
        .try_into()
        .unwrap();
    let upper_bytes: [u8; 16] = account_data[200..216]
        .try_into()
        .unwrap();
    
    let lower_sqrt_price = u128::from_le_bytes(lower_bytes);
    let upper_sqrt_price = u128::from_le_bytes(upper_bytes);
    
    Ok((lower_sqrt_price, upper_sqrt_price))
}

// Usage
let account = rpc_client.get_account(&position_pda)?;
let (lower, upper) = parse_position_limits(&account.data)?;

// Convert to human-readable prices
let lower_price = sqrt_price_to_price(lower, 9, 6);
let upper_price = sqrt_price_to_price(upper, 9, 6);

println!("Buy limit: ${:.2}", lower_price);
println!("Sell limit: ${:.2}", upper_price);

4. Watching for Transaction Confirmation​

use solana_sdk::signature::Signature;
use std::time::Duration;

fn wait_for_confirmation(
    rpc_client: &RpcClient,
    signature: &Signature,
    max_retries: u32,
) -> Result<()> {
    for i in 0..max_retries {
        std::thread::sleep(Duration::from_secs(2));
        
        if let Ok(status) = rpc_client.get_signature_status(signature) {
            if let Some(result) = status {
                if let Err(e) = result {
                    return Err(anyhow!("Transaction failed: {:?}", e));
                }
                println!("✅ Transaction confirmed in slot");
                return Ok(());
            }
        }
        
        if i == max_retries - 1 {
            return Err(anyhow!("Transaction timeout"));
        }
    }
    
    Ok(())
}

Advanced: Modify Position with Collateral​

The modify_tuna_spot_position_orca instruction adds collateral and activates trading:

// From IDL
pub struct ModifyTunaSpotPositionOrca {
    pub decrease_percent: u32,      // 0 for increase
    pub collateral_amount: u64,     // USDC to deposit
    pub borrow_amount: u64,         // SOL to borrow from vault
    pub required_swap_amount: u64,  // 0 for auto-calc
    pub remaining_accounts_info: RemainingAccountsInfo,
}

// Required accounts (24 total)
accounts: vec![
    AccountMeta::new(authority, true),
    AccountMeta::new_readonly(tuna_config, false),
    AccountMeta::new_readonly(mint_a, false),
    AccountMeta::new_readonly(mint_b, false),
    AccountMeta::new_readonly(token_program_a, false),
    AccountMeta::new_readonly(token_program_b, false),
    AccountMeta::new(market, false),
    AccountMeta::new(vault_a, false),
    AccountMeta::new(vault_b, false),
    AccountMeta::new(vault_a_ata, false),
    AccountMeta::new(vault_b_ata, false),
    AccountMeta::new(tuna_spot_position, false),
    AccountMeta::new(tuna_position_ata_a, false),
    AccountMeta::new(tuna_position_ata_b, false),
    AccountMeta::new(authority_ata_a, false),
    AccountMeta::new(authority_ata_b, false),
    AccountMeta::new(fee_recipient_ata_a, false),
    AccountMeta::new(fee_recipient_ata_b, false),
    AccountMeta::new_readonly(pyth_oracle_price_feed_a, false),
    AccountMeta::new_readonly(pyth_oracle_price_feed_b, false),
    AccountMeta::new_readonly(whirlpool_program, false),
    AccountMeta::new(whirlpool, false),
    AccountMeta::new_readonly(memo_program, false),
    AccountMeta::new_readonly(system_program::ID, false),
]

Why so many accounts?

• DeFiTuna needs to interact with lending vaults
• Price oracles (Pyth) for health checks
• Fee collection accounts
• Orca Whirlpool state for actual swaps
• Multiple token accounts for each token in the pair

Price Calculation Mathematics​

Square Root Price Encoding​

DeFiTuna stores prices as $\sqrt{P} \cdot 2^{64}$ (same as Uniswap V3):