Implementation of new copying service

[loboda4450]

Service reacts to a webhook when a new file shows up. It pulls the file, checks with the existing dedup service to avoid copying duplicates, and only continues if the file is new. If the file is a ZIP, the service looks inside and decides which files should be copied and which should be skipped based on some rules (tbd). The goal is to sync only what actually matters, without duplicating data or copying unnecessary stuff.

Desirably, if an new album is downloaded and there are currently imported tracks from that particular album it should replace currently existing ones with the album ones to maintain filesystem as clean as possible, but this produces another challenges as to maintain database integrity (favourites/play count/etc) of streaming services.

[JuniorJPDJ]

/opencode how would you plan this?

[github-actions]

Based on the codebase analysis, here's my implementation plan:

Architecture Overview

┌─────────────────┐     ┌─────────────┐     ┌──────────────┐
│  Webhook Source │────▶│ Copy Service│────▶│ Dedup Service│
└─────────────────┘     └─────────────┘     └──────────────┘
                              │
                              ▼
                       ┌──────────────┐
                       │ File Storage │
                       └──────────────┘

Implementation Plan

1. New Service Module: copy_service.py

• FastAPI service similar to connector.py
• Webhook endpoint for new file notifications
• Orchestrates pull → dedup check → copy workflow

2. ZIP Processing Module: zip_handler.py

• Extract ZIP contents to temp directory
• Filter files based on rules (extensions, metadata, etc.)
• Return list of files to copy

3. Copy Engine: copy_engine.py

• Pull files from source (HTTP/S3/local)
• Query dedup service (via API or direct DB) to check duplicates
• Copy non-duplicate files to destination
• Handle album replacement logic

4. Database Schema Updates

Add to lib/sqlmodels.py:

class CopyJob(SQLBase):
    """Track copy operations"""
    id = Column(Integer, primary_key=True)
    source_path = Column(String, nullable=False)
    dest_path = Column(String, nullable=False)
    status = Column(SQLEenum(CopyJobStatus))
    album = Column(String)  # For album replacement tracking

5. Album Replacement Logic

• When full album downloaded: query existing tracks by (album, mb_albumid)
• Mark old tracks for deprecation but keep path reference
• Preserve metadata integrity (favourites, play count from streaming service DB)
• New design: Add replaced_by foreign key to Track table

6. Key Design Decisions

Dedup Integration Options:

• Option A: Direct DB access (simpler, faster)
• Option B: HTTP API to dedup service (decoupled)
• Recommendation: Option A for performance, with clear separation in code

ZIP Filtering Rules:

• Audio file extensions (.mp3, .flac, .m4a)
• Skip based on metadata rules (e.g., skip explicit versions if clean exists)
• File size thresholds

Album Replacement Strategy:

• Add is_active flag to Track model
• When replacing: old tracks is_active=False, new is_active=True
• Path history maintained in new CopyJob table

7. File Structure

copy_service.py          # Main FastAPI service
copy_engine.py           # Copy orchestration logic
zip_handler.py           # ZIP extraction & filtering
requirements.txt          # Add: zipfile, boto3 (if S3)

8. API Endpoints

POST /copy_job            # Receive webhook, start copy job
GET  /copy_job/{id}       # Job status
GET  /copy_jobs           # List jobs

9. Configuration

• --source-dir: Source file location
• --dest-dir: Destination for copied files
• --dedup-db: Shared dedup database
• --webhook-url: Notification endpoint on completion

10. Workflow Sequence

1. Webhook → POST /copy_job
2. Pull file (or download from URL)
3. If ZIP: extract & filter
4. For each file:
   • Generate acoustic fingerprint
   • Query dedup DB for duplicates
   • If not duplicate: copy to dest
   • If duplicate: skip or handle replacement
5. Update dedup DB with new tracks
6. Send completion webhook

11. Challenges to Address

• Metadata preservation: Need integration points with streaming services (Navidrome/Jellyfin/Airsonic)
• Concurrency: Multiple copy jobs handling same album
• Error recovery: Partial copy scenarios
• Performance: Batch fingerprinting for large ZIPs

Would you like me to proceed with implementing this plan?

opencode session  |  github run

[JuniorJPDJ]

This is tgmount webhook from github juniorjpdj/tgmount project. Example of handling that webhook is dedup tgmount endpoint. Let's start with simple stuff - connector tgmount copy webhook that will copy file to argument defined directory preserving metadata like mtime, mode, etc. /oc update and simplify the plan. Graphs should be outputted in mermaid format and outputed in github-compatible format.

[github-actions]

Based on the codebase analysis, here's the simplified implementation plan for the tgmount copy webhook service:

Simplified Architecture

flowchart LR
    A[Telegram] --> B[tgmount]
    B -->|webhook| C[Copy Connector]
    C --> D[Copy to Destination]
    
    subgraph Existing Services
        E[Dedup Worker]
        F[Connector]
    end

Loading

Implementation Plan

1. New Service: copy_connector.py

A simple FastAPI service that:

• Receives tgmount webhook with new file path
• Copies file to destination directory preserving metadata
• Optional: queries dedup DB to check for duplicates (future enhancement)

2. Key Features

Minimal V1:

• FastAPI webhook endpoint (/tgmount_copy)
• Copy file preserving mtime, mode, atime
• Optional async queue support (same pattern as dedup connector)

Database (Optional for V1):
Can operate without DB initially. If needed, add:

class CopyJob(SQLBase):
    id = Column(Integer, primary_key=True)
    source_path = Column(String, nullable=False)
    dest_path = Column(String, nullable=False)
    status = Column(SQLEnum(CopyJobStatus))
    created_at = Column(DateTime, default=func.now())

3. File Structure

copy_connector.py          # Main FastAPI webhook service (NEW)

4. Configuration

--basedir              # Source directory (tgmount mount)
--dest-dir             # Destination directory for copies
--host                 # Listen address (default: 0.0.0.0)
--port                 # Listen port (default: 8001)
--db                   # Optional: database URL for job tracking

5. API Endpoint

POST /tgmount_copy
Request body: {"fname": "path/to/file.ext"}
Response: {"status": "ok", "source": "...", "dest": "..."}

6. Data Flow

sequenceDiagram
    participant T as tgmount
    participant C as Copy Connector
    participant D as Destination Dir
    
    T->>C: POST /tgmount_copy {"fname": "file.mp3"}
    C->>C: Resolve full path from basedir
    C->>D: Copy file preserving metadata
    C->>T: Return {"status": "ok"}

Loading

7. Docker Compose Integration

copy_connector:
  image: ghcr.io/fumr/flaczkownia-dedup
  restart: unless-stopped
  volumes:
    - source: ./data/tgmount/mnt
      target: /app/source
      read_only: true
      type: bind
    - source: ./data/copied
      target: /app/dest
      type: bind
  command:
    - python
    - copy_connector.py
    - --basedir
    - /app/source/flaczkownia
    - --dest-dir
    - /app/dest

8. Implementation Details

File Copy with Metadata Preservation:

import shutil
import os

def copy_with_metadata(src, dst):
    shutil.copy2(src, dst)  # Preserves mtime, mode, atime

Webhook Model:

class TGMountWebhook(BaseModel):
    fname: str

9. Future Enhancements (V2+)

• Add dedup DB query to skip duplicates
• Add ZIP file extraction and filtering
• Add album replacement logic
• Add job queue with status tracking

10. Simplified Flow Diagram

flowchart TD
    A[New File in tgmount] --> B{Webhook Triggered}
    B --> C[Copy Connector]
    C --> D[Resolve Source Path]
    D --> E{Optional: Check Dedup DB}
    E -->|No DB| F[Copy File]
    E -->|DB Check| G{Duplicate?}
    G -->|Yes| H[Skip Copy]
    G -->|No| F
    F --> I[Preserve Metadata]
    I --> J[Save to Destination]
    J --> K[Return Success]

Loading

11. Minimal Changes Required

• New file: copy_connector.py (~100 lines)
• No database schema changes for V1 (optional for V2)
• No changes to existing services
• No new dependencies (uses only stdlib + FastAPI)

This provides a simple, working copy service that can be enhanced incrementally based on actual usage patterns.

opencode session  |  github run

[JuniorJPDJ]

/oc Analyze what the whole project does right now please

[github-actions]

Project Analysis

Flaczkownia Dedup is a music file deduplication system that prevents duplicate tracks from being imported into self-hosted streaming services.

Current Architecture

flowchart LR
    A[Telegram] --> B[tgmount]
    B -->|new file webhook| C[Connector]
    C --> D[Queue DB]
    E[ziprofs] --> F[Dedup Worker]
    F -->|process files| D
    F -->|results webhook| C
    C -->|symlinks| G[Deduped View]

Loading

Components

dedup.py - Core deduplication worker:

• Generates acoustic fingerprints using audioprint
• Deduplicates based on tuple: (album_name, disc_number, track_number, acoustic_hash)
• Runs as daemon (queue mode) or one-shot directory processor
• Sends webhooks on file completion

connector.py - FastAPI service:

• POST /tgmount_add_to_dedup_queue - receives tgmount webhooks
• POST /dedup_processed_file_webhook - receives dedup results
• Creates symlink view of non-duplicate files
• Manages stale symlink cleanup

backfill_queue.py - Populates queue with existing files

Database (SQLite):

• Queue - pending dedup jobs
• Track - track metadata + duplicate flag
• UnknownFile - unsupported file formats

Docker Compose - orchestrates: tgmount (Telegram → FS), ziprofs (ZIP → FS), dedup (worker), connector (API)

opencode session  |  github run

[JuniorJPDJ]

Ok, so using the plan from previous comment imagine that we want to add additional step that copy files to local directory before running deduplication. /oc

[github-actions]

Author identity unknown

*** Please tell me who you are.

Run

git config --global user.email "you@example.com"
git config --global user.name "Your Name"

to set your account's default identity.
Omit --global to set the identity only in this repository.

fatal: empty ident name (for runner@runnervmkj6or.g3hwm4zdrdxutl2hw05zhfo4lf.dx.internal.cloudapp.net) not allowed

opencode session  |  github run

[JuniorJPDJ]

Ok, so using the plan from previous comment imagine that we want to add additional step that copy files to local directory before running deduplication. The step should be handled in connector. Show me the detailed plan. Graphs should be outputted in mermaid and github-compatible format. /oc

[github-actions]

Updated Architecture with Copy Step

Current vs New Flow

flowchart LR
    subgraph Current Flow
        A1[Telegram] --> B1[tgmount]
        B1 -->|webhook| C1[Connector]
        C1 --> D1[Queue DB]
        E1[ziprofs] --> F1[Dedup Worker]
        F1 -->|results| C1
        C1 --> G1[Deduped View]
    end
    
    subgraph New Flow
        A2[Telegram] --> B2[tgmount]
        B2 -->|webhook| C2[Connector]
        C2 --> H2[Copy to Local Dir]
        H2 --> D2[Queue DB]
        E2[ziprofs] --> F2[Dedup Worker]
        F2 -->|results| C2
        C2 --> G2[Deduped View]
    end

Loading

Detailed Sequence

sequenceDiagram
    participant T as tgmount
    participant C as Connector
    participant L as Local Copy Dir
    participant Q as Queue DB
    participant W as Dedup Worker
    participant V as Deduped View
    
    T->>C: POST /tgmount_add_to_dedup_queue
    Note over C: Receive file path
    
    C->>L: Copy file preserving metadata
    Note over L: File stored in local directory
    
    C->>Q: Add job to queue
    Note over Q: Job references local copy
    
    W->>Q: Get next job
    W->>L: Read file from local dir
    W->>W: Process deduplication
    W->>C: POST /dedup_processed_file_webhook
    
    C->>V: Create symlink to local copy

Loading

Connector Flow Diagram

flowchart TD
    A[POST /tgmount_add_to_dedup_queue] --> B{Source is remote?}
    B -->|No - local| C[Copy to Local Dir]
    B -->|Yes - already local| D[Skip Copy]
    
    C --> E[Preserve mtime, mode, atime]
    E --> F[Generate new local path]
    F --> G{Copy success?}
    
    G -->|No| H[Return Error]
    G -->|Yes| I[Add to Queue DB]
    
    D --> I
    
    I --> J[Queue Job]
    J --> K[Return Success]
    
    style C fill:#90EE90
    style E fill:#90EE90
    style F fill:#90EE90

Loading

Implementation Changes

1. Update connector.py

New Configuration:

--local-copy-dir    # Directory to store copied files (optional)

Add Copy Function:

def copy_to_local_dir(source_path: str, local_copy_dir: str) -> str:
    """Copy file to local directory preserving metadata"""
    os.makedirs(local_copy_dir, exist_ok=True)
    
    # Generate unique filename
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
    basename = os.path.basename(source_path)
    dest_path = os.path.join(local_copy_dir, f"{timestamp}_{basename}")
    
    shutil.copy2(source_path, dest_path)  # Preserves mtime, mode, atime
    return dest_path

Update Webhook Handler:

@app.post("/tgmount_add_to_dedup_queue")
async def add_to_dedup_queue(body: TGMountWebhook):
    full_path = resolve_basedir_path(args.basedir, body.fname)
    
    if args.local_copy_dir:
        # NEW: Copy file to local directory first
        full_path = copy_to_local_dir(full_path, args.local_copy_dir)
    
    # Add to queue
    add_to_queue(full_path, body.fname)
    return {"status": "ok"}

2. Database Schema (No Changes)

Queue job will reference the local copy path after copy step.

3. Docker Compose Updates

connector:
  image: ghcr.io/fumr/flaczkownia-dedup
  restart: unless-stopped
  volumes:
    - source: ./data/tgmount/mnt
      target: /app/tgmount
      read_only: true
      type: bind
    - source: ./data/local_copy
      target: /app/local_copy
      type: bind
    - source: ./data/deduped
      target: /app/deduped
      type: bind
  command:
    - python
    - connector.py
    - --basedir
    - /app/tgmount/flaczkownia
    - --deduped-dir
    - /app/deduped
    - --local-copy-dir
    - /app/local_copy  # NEW

dedup_worker:
  image: ghcr.io/fumr/flaczkownia-dedup
  restart: unless-stopped
  volumes:
    - source: ./data/local_copy
      target: /app/local_copy
      read_only: true
      type: bind  # Changed: now reads from local copy dir
  # ... other config

4. Benefits

1. Separation of concerns: Original files remain untouched
2. Atomic operations: Copy happens before dedup, ensuring data availability
3. Backup capability: Original files preserved in tgmount
4. Simpler cleanup: Local copy dir can be pruned independently
5. Metadata preservation: shutil.copy2 maintains all file metadata

5. Error Handling

flowchart TD
    A[Webhook Received] --> B{local-copy-dir set?}
    B -->|No| E[Use original path]
    B -->|Yes| C[Attempt Copy]
    
    C --> D{Copy Success?}
    D -->|No| F[Return Error 500]
    D -->|Yes| G[Use local copy path]
    
    E --> H[Add to Queue]
    G --> H
    H --> I[Return Success]

Loading

6. Minimal Code Changes

connector.py modifications (~30 lines):

• Add --local-copy-dir argument
• Add copy_to_local_dir() function
• Update /tgmount_add_to_dedup_queue endpoint
• Add error handling for copy failures

No changes needed to:

• dedup.py (reads from queue path)
• Database schema
• Other services

This approach keeps the changes minimal while adding the copy functionality to the connector workflow.

opencode session  |  github run