[FEATURE] Automatic Bucket Processing with User Access Controls

[davidamacey]

Automatic Bucket Processing and User Access Control

Problem Statement

Users need to upload files directly to MinIO for automatic processing, with proper access controls.

Solution

Implement a bucket watcher service with user access controls.

1. Bucket Watcher Service

New Module: app/services/bucket_watcher.py

from minio import Minio
from minio.notification import NotificationConfig
from sqlalchemy.orm import Session
import logging

class BucketWatcher:
    def __init__(self):
        self.bucket_name = settings.MEDIA_BUCKET_NAME
        self.input_prefix = "auto-ingest/"
        
    def start_watching(self):
        config = NotificationConfig(
            queue_config_list=[
                NotificationConfig.QueueConfig(
                    "arn:minio:sqs::1:webhook",
                    ["s3:ObjectCreated:*"],
                    prefix=f"{self.input_prefix}user-",
                    suffix=".mp3,.wav,.m4a,.mp4,.mov"
                )
            ]
        )
        minio_client.set_bucket_notification(self.bucket_name, config)

2. User Access Controls

New Model: app/models/access.py

from sqlalchemy import Column, Integer, ForeignKey, Boolean, DateTime
from sqlalchemy.orm import relationship

class UserBucket(Base):
    __tablename__ = "user_buckets"
    
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
    bucket_name = Column(String, unique=True, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=func.now())
    
    user = relationship("User", back_populates="buckets")

3. API Endpoints

New Module: app/api/endpoints/buckets.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

router = APIRouter()

@router.post("/buckets/")
def create_bucket(
    bucket_name: str,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
):
    """Create a new user bucket"""
    if db.query(UserBucket).filter(UserBucket.bucket_name == bucket_name).first():
        raise HTTPException(400, "Bucket name already exists")
        
    bucket = UserBucket(
        user_id=current_user.id,
        bucket_name=bucket_name
    )
    
    db.add(bucket)
    db.commit()
    return {"message": "Bucket created"}

4. Configuration

Add to .env:

AUTO_INGEST_BUCKET=opentranscribe-ingest
AUTO_INGEST_PREFIX=uploads/

5. Deployment

Update docker-compose.yml:

services:
  bucket-watcher:
    build: ./backend
    command: python -m app.services.bucket_watcher
    environment:
      - MINIO_HOST=minio
      - MINIO_PORT=9000
    depends_on:
      - minio
      - redis

Testing Plan

1. Unit Tests

   • Test bucket creation
   • Test file processing
   • Test access controls

2. Integration Tests

   • Test complete upload flow
   • Test permission checks

3. Manual Testing

   • Upload files via MinIO client
   • Verify processing
   • Check access controls

Related Issues

• [FEATURE] Cross-Platform PyTorch Model Support for CUDA, MPS, and CPU #25 - Cross-Platform PyTorch Support

[davidamacey]

Comprehensive Implementation Plan - SMB Folder Watcher with Email Notifications

We've developed a detailed implementation plan that extends the original bucket watcher concept to support SMB/CIFS network shares with automatic file ingestion and email notifications when transcription completes.

Key Features Planned

1. SMB Watch Folder Monitoring

• Poll-based watching at configurable intervals (default: 5 minutes)
• Credential storage using existing Fernet encryption (same pattern as LLM API keys)
• File filtering with include/exclude glob patterns
• Recursive subfolder scanning option
• Connection testing before saving configuration

2. Fast Duplicate Detection (imohash)

• Uses imohash library which samples strategic portions of files (first 16KB, middle, last 16KB)
• Extremely fast for large media files vs full MD5/SHA256
• Prevents re-processing of already completed files
• Detects file modifications for reprocessing

3. Multi-Part File Stitching

• Detects split recordings with naming pattern: filename_P001.mp4, filename_P002.mp4, etc.
• Automatically stitches parts with ffmpeg (stream copy, no re-encoding)
• Uploads stitched file back to SMB share
• Processes combined file as single recording
• Handles broken VTC/podcast recordings from connection issues

4. Multi-Provider Email Notifications

• SMTP - Gmail, Yahoo, self-hosted mail servers
• Microsoft 365 - OAuth2 with Azure AD App Registration + Microsoft Graph API (for corporate/government environments that disable basic auth)
• Exchange On-Premises - Legacy support with SMTP/NTLM
• Email includes: BLUF summary, speaker analysis, action items, file metadata

Architecture Flow

User configures SMB Watch Folder (Settings UI)
           ↓
Celery Beat triggers scan every minute
           ↓
SMB Service connects, lists files, calculates imohash
           ↓
Check for duplicates (hash match = skip)
           ↓
Detect multi-part files → Stitch with ffmpeg if complete group
           ↓
Download to MinIO → Transcription pipeline
           ↓
On completion: Email notification (if configured)

New Database Tables

1. smb_watch_folder - Watch folder configurations with encrypted credentials
2. smb_processed_file - Track processed files (hash, status, multi-part info)
3. email_notification_config - Multi-provider email settings (SMTP/Microsoft 365/Exchange)
4. smb_watch_folder_email - Junction table linking folders to email configs

New Dependencies

• smbprotocol>=1.12.0 - Pure Python SMB client (cross-platform)
• imohash>=1.0.4 - Fast file hashing
• msal>=1.26.0 - Microsoft Authentication Library (for M365 OAuth2)

Security

• All credentials (SMB passwords, SMTP passwords, Azure secrets) encrypted with Fernet
• Never exposed in API responses (uses has_password: bool pattern)
• Follows existing FedRAMP-compliant patterns in codebase

---

📄 Full Implementation Plan: https://gist.github.com/davidamacey/6fc7a2b36a9d74a7811e3a13b685bbb8

The plan includes:

• Complete database schemas
• SQLAlchemy models and Pydantic schemas
• SMB service implementation
• Email service with multi-provider support
• Celery task designs
• API endpoint specifications
• Frontend component structure
• Testing and verification approach

---

Generated with Claude Code planning session

[davidamacey]

Updated Implementation Plan: Watch Source Monitoring (SMB + S3) & Email Notifications

Building on the original gist, the scope has been expanded to include S3-compatible bucket watching alongside SMB. Both source types use a unified architecture with a common processing pipeline.

Full implementation plan (gist):

https://gist.github.com/davidamacey/1d5a5c6fa159fa0d873fe41a3775dbc3

Key Design Decisions

1. Unified Watch Source Architecture
Rather than two separate systems, we use a single watch_source database table with a source_type discriminator column (smb | s3), and source-specific credential/config columns. A common abstract BaseWatchSourceClient interface is implemented by SMBClient (smbprotocol) and S3Client (boto3). A factory function creates the appropriate client based on source_type. This keeps the Celery tasks, processing pipeline, email integration, and UI unified.

2. Admin-Only Access
All watch source configuration is restricted to admins. Imported files are assigned to a configurable user via assign_to_user_id. Non-admin users see imported files in their gallery but cannot manage watch sources.

3. Skip Files Older Than N Days
Each watch source has a skip_files_older_than_days field (default 30, nullable to disable). Files with modified_time older than now - N days are recorded as skipped_old in the processed files table and not downloaded.

4. Imohash Duplicate Detection
Uses imohash which reads only 48KB per file (first 16KB + middle 16KB + last 16KB + file size) regardless of file size. Stored in watch_source_processed_file.imohash with a composite index on (watch_source_id, imohash).

5. Polling Architecture
Celery Beat runs a lightweight orchestrator task every 5 minutes. Each watch source's polling_interval_minutes is checked against last_scan_at inside the beat task. Only sources that are due get dispatched for scanning. This avoids dynamic beat schedule modification, which is fragile in Celery.

6. S3 Testing
The existing MinIO instance doubles as the S3-compatible test endpoint. A separate watch-source-test bucket (distinct from the internal transcriptions bucket) is created by the test data setup script.

7. Encrypted Credentials
All SMB passwords, S3 secret access keys, and email provider credentials are encrypted at rest using the existing AES-256-GCM encryption (encrypt_api_key() / decrypt_api_key() from app.utils.encryption). API responses use has_password: bool computed properties and never expose encrypted values.

Source Types

Source	Library	Auth	Use Case
SMB/CIFS	smbprotocol (pure Python)	Username/password/domain	Windows shared drives (gov/corporate)
S3	boto3	Access key + secret key	AWS S3, MinIO, Backblaze B2, Wasabi, any S3-compatible

New Dependencies

• smbprotocol>=1.12.0 — Pure Python SMB2/3 client
• imohash>=1.0.4 — Fast partial-file hashing
• msal>=1.26.0 — Microsoft Auth Library (M365 email)
• boto3>=1.34.0 — AWS S3 SDK (check if already present)

Implementation Phases (9 total)

1. Database schema + models + schemas (4 tables)
2. Storage source services (abstract base + SMB + S3 clients)
3. File processing service (imohash + multi-part stitching)
4. Email service (SMTP + M365 + Exchange)
5. API endpoints (CRUD + testing + scanning)
6. Celery tasks (periodic scan + file processing + email notification)
7. Frontend settings UI (unified panel with source type tabs)
8. Docker test container + integration (SMB container + MinIO S3 bucket)
9. Tests

See the full plan gist for detailed file lists, schemas, and verification steps.

[davidamacey]

Implementation Plan Ready

A comprehensive implementation plan has been developed for this feature, significantly expanding the original scope to cover:

• Watch Sources: Local folder (Docker volume mount), S3, and SMB with configurable scan intervals
• Dual-Hash Deduplication: New backend imohash field (64KB sampling, SHA-256) for robust cross-pipeline dedup, alongside existing frontend file_hash for quick UI checks
• Multi-Part Stitching: Auto-detects _P### suffixed files from the same recording session (lost connections), groups by time window, validates codec compatibility via ffprobe, and stitches with ffmpeg concat
• Local Folder Browser: UI folder picker that dynamically lists directories under the Docker mount point
• S3/SMB Copy-to-MinIO: Files are copied into MinIO for processing; originals are never moved or deleted from remote sources
• Imohash Backfill Migration: Existing files get backend imohash computed via Celery migration task (follows thumbnail/embedding migration pattern)
• 3-Layer Deduplication: Path-level within source → hash-level across sources → cross-pipeline via media_file.imohash

Implementation Phases

Phase	Scope
0	Backend imohash foundation (new field, algorithm, migration, integration into upload/URL pipelines)
1	Database schema, models, encrypted credentials for watch sources
2	Core services (scanner, importer, stitcher, folder browser, S3/SMB connectors)
3	Celery tasks, beat schedule, scan orchestration
4	Frontend UI (source management, folder picker, status dashboard)
5	API endpoints, WebSocket events, Docker compose overlay
6	Polish (edge cases, error recovery, tests)

29 new files, 19 modified files across all phases.

Full Plan

📋 Watch Source Auto-Import Implementation Plan (~1200 lines, covers all edge cases, file listings, and verification plan)

This supersedes the earlier gist plans linked in the issue description, incorporating all feedback and architectural decisions.