improve work

app/core/background/convert_service.py

@@ -136,8 +136,7 @@ async def convert_loop(memory):
             ]
             if trim_value:
                 cmd.extend(["--trim", trim_value])
-            if content_kind == "audio":
-                cmd.append("--audio-only")  # audio-only flag
+            # converter auto-detects audio/video, no explicit flag required
 
             process = await asyncio.create_subprocess_exec(
                 *cmd,

app/core/background/convert_v3_service.py

@@ -2,8 +2,11 @@ import asyncio
 import os
 import json
 import shutil
+import tempfile
+from dataclasses import dataclass
 from datetime import datetime
-from typing import List, Tuple, Optional
+from pathlib import Path
+from typing import List, Optional, Tuple
 
 from sqlalchemy import select
 
@@ -22,9 +25,41 @@ from app.core.crypto.encf_stream import decrypt_encf_auto
 from app.core.crypto.keywrap import unwrap_dek, wrap_dek, KeyWrapError
 from app.core.network.key_client import request_key_from_peer
 from app.core.models.my_network import KnownNode
+from app.core._utils.resolve_content import resolve_content
+from app.core.content.content_id import ContentId
 
 
 CONCURRENCY = int(os.getenv("CONVERT_V3_MAX_CONCURRENCY", "3"))
+STAGING_SUBDIR = os.getenv("CONVERT_V3_STAGING_SUBDIR", "convert-staging")
+UPLOADS_PATH = Path(UPLOADS_DIR).resolve()
+_host_uploads_env = os.getenv("BACKEND_DATA_DIR_HOST")
+HOST_UPLOADS_PATH = Path(_host_uploads_env).resolve() if _host_uploads_env else None
+
+
+@dataclass
+class PlainStaging:
+    container_path: str
+    host_path: str
+
+
+def _container_to_host(path: str) -> str:
+    """Map a container path under UPLOADS_DIR to the host path for docker -v."""
+    if not HOST_UPLOADS_PATH:
+        raise RuntimeError("BACKEND_DATA_DIR_HOST is not configured for convert_v3")
+    real_path = Path(path).resolve()
+    try:
+        real_path.relative_to(UPLOADS_PATH)
+    except ValueError:
+        # Not under uploads; best effort fallback to original string
+        return str(real_path)
+    rel = real_path.relative_to(UPLOADS_PATH)
+    return str(HOST_UPLOADS_PATH / rel)
+
+
+MEDIA_CONVERTER_CPU_LIMIT = os.getenv("MEDIA_CONVERTER_CPU_LIMIT")
+MEDIA_CONVERTER_MEM_LIMIT = os.getenv("MEDIA_CONVERTER_MEM_LIMIT")
+MEDIA_CONVERTER_CPUSET = os.getenv("MEDIA_CONVERTER_CPUSET") or os.getenv("CONVERT_CPUSET")
+ERROR_TRUNCATE_LIMIT = 512
 
 
 def _ensure_dir(path: str):
@@ -57,27 +92,49 @@ async def _save_derivative(file_path: str, filename: str) -> Tuple[str, int]:
     return file_hash, size
 
 
-async def _run_media_converter(input_host_path: str, input_ext: str, quality: str, trim_value: Optional[str], is_audio: bool) -> Tuple[str, dict]:
+async def _run_media_converter(staging: PlainStaging, input_ext: str, quality: str, trim_value: Optional[str], is_audio: bool):
+    if not os.path.exists(staging.container_path):
+        raise FileNotFoundError(f"Plain input missing at {staging.container_path}")
+
+    host_input_path = staging.host_path
+    if not host_input_path or not host_input_path.startswith('/'):
+        host_input_path = os.path.abspath(host_input_path)
+
     rid = __import__('uuid').uuid4().hex[:8]
-    output_dir_container = f"/tmp/conv_{rid}"
-    output_dir_host = f"/tmp/conv_{rid}"
-    _ensure_dir(output_dir_host)
-    logs_dir_host = BACKEND_LOGS_DIR_HOST
-    _ensure_dir(logs_dir_host)
+    output_dir_container = UPLOADS_PATH / "convert-output" / f"conv_{rid}"
+    output_dir_host = _container_to_host(output_dir_container)
+    _ensure_dir(str(output_dir_container))
+
+    logs_dir_candidate = os.getenv("BACKEND_LOGS_DIR_HOST", "")
+    logs_dir_host = logs_dir_candidate if logs_dir_candidate else str(HOST_UPLOADS_PATH / "logs" / "converter") if HOST_UPLOADS_PATH else "/tmp/converter-logs"
+    if not logs_dir_host.startswith('/'):
+        logs_dir_host = os.path.join(os.getcwd(), logs_dir_host)
+    try:
+        os.makedirs(logs_dir_host, exist_ok=True)
+    except Exception:
+        fallback_logs = HOST_UPLOADS_PATH / "logs" / "converter" if HOST_UPLOADS_PATH else Path("/tmp/converter-logs")
+        logs_dir_host = str(fallback_logs)
+        os.makedirs(logs_dir_host, exist_ok=True)
 
     cmd = [
         "docker", "run", "--rm",
-        "-v", f"{input_host_path}:/app/input:ro",
+        "-v", f"{host_input_path}:/app/input:ro",
         "-v", f"{output_dir_host}:/app/output",
         "-v", f"{logs_dir_host}:/app/logs",
-        "media_converter",
-        "--ext", input_ext,
-        "--quality", quality,
     ]
+    if MEDIA_CONVERTER_CPU_LIMIT:
+        cmd.extend(["--cpus", str(MEDIA_CONVERTER_CPU_LIMIT)])
+    if MEDIA_CONVERTER_MEM_LIMIT:
+        cmd.extend(["--memory", str(MEDIA_CONVERTER_MEM_LIMIT)])
+    if MEDIA_CONVERTER_CPUSET:
+        cmd.extend(["--cpuset-cpus", MEDIA_CONVERTER_CPUSET])
+
+    cmd.append("media_converter")
+    cmd.extend(["--ext", input_ext, "--quality", quality])
     if trim_value:
         cmd.extend(["--trim", trim_value])
-    if is_audio:
-        cmd.append("--audio-only")
+
+    make_log('convert_v3', f"Run media_converter cmd: {' '.join(cmd)}")
 
     proc = await asyncio.create_subprocess_exec(
         *cmd,
@@ -90,15 +147,15 @@ async def _run_media_converter(input_host_path: str, input_ext: str, quality: st
 
     # Find produced media file and optional output.json
     try:
-        files = os.listdir(output_dir_host)
+        files = os.listdir(output_dir_container)
     except Exception as e:
         raise RuntimeError(f"Read output dir error: {e}")
     media_files = [f for f in files if f != "output.json"]
     if len(media_files) != 1:
         raise RuntimeError(f"Expected one media file, found {len(media_files)}: {media_files}")
-    output_media = os.path.join(output_dir_host, media_files[0])
+    output_media = os.path.join(output_dir_container, media_files[0])
     ffprobe_meta = {}
-    out_json = os.path.join(output_dir_host, "output.json")
+    out_json = os.path.join(output_dir_container, "output.json")
     if os.path.exists(out_json):
         try:
             with open(out_json, 'r') as f:
@@ -108,24 +165,74 @@ async def _run_media_converter(input_host_path: str, input_ext: str, quality: st
     return output_media, ffprobe_meta
 
 
-async def _convert_content(ec: EncryptedContent, input_host_path: str):
-    content_kind = 'audio' if ec.content_type.startswith('audio/') else ('video' if ec.content_type.startswith('video/') else 'other')
-    if content_kind == 'other':
-        return
+async def _update_upload_session(ec: EncryptedContent, all_success: bool, errors: List[str]):
+    async with db_session() as session:
+        upload_row = (await session.execute(
+            select(UploadSession).where(UploadSession.encrypted_cid == ec.encrypted_cid)
+        )).scalars().first()
+        if upload_row:
+            if all_success:
+                upload_row.state = 'converted'
+                upload_row.error = None
+            elif upload_row.state != 'converted':
+                upload_row.state = 'conversion_failed'
+                if errors:
+                    upload_row.error = _short_error(errors[0])
+            await session.commit()
 
+
+async def _convert_content(ec: EncryptedContent, staging: PlainStaging):
+    content_kind = 'audio' if ec.content_type.startswith('audio/') else ('video' if ec.content_type.startswith('video/') else 'other')
     input_ext = (ec.content_type.split('/')[-1] or 'bin')
     is_audio = content_kind == 'audio'
-    # Required outputs
-    required = ['high', 'low', 'low_preview']
+    encrypted_hash_b58 = ContentId.deserialize(ec.encrypted_cid).content_hash_b58
 
-    # Preview interval
+    if content_kind == 'other':
+        errors: List[str] = []
+        all_success = True
+        try:
+            file_hash, size_bytes = await _save_derivative(staging.container_path, staging.container_path)
+            plain_path = os.path.join(UPLOADS_DIR, file_hash)
+            plain_filename = f"{ec.encrypted_cid}.{input_ext}" if input_ext else ec.encrypted_cid
+            async with db_session() as session:
+                existing = (await session.execute(select(StoredContent).where(StoredContent.hash == file_hash))).scalars().first()
+                if not existing:
+                    sc = StoredContent(
+                        type="local/content_bin",
+                        hash=file_hash,
+                        user_id=None,
+                        filename=plain_filename,
+                        meta={'encrypted_cid': ec.encrypted_cid, 'kind': 'original'},
+                        created=datetime.utcnow(),
+                    )
+                    session.add(sc)
+                    await session.flush()
+                derivative = ContentDerivative(
+                    content_id=ec.id,
+                    kind='decrypted_original',
+                    local_path=plain_path,
+                    content_type=ec.content_type,
+                    size_bytes=size_bytes,
+                    status='ready',
+                )
+                session.add(derivative)
+                await session.commit()
+            make_log('convert_v3', f"Stored original derivative for {ec.encrypted_cid}")
+        except Exception as e:
+            all_success = False
+            errors.append(str(e))
+            make_log('convert_v3', f"Convert error {ec.encrypted_cid} opt=original: {e}", level='error')
+        await _update_upload_session(ec, all_success, errors)
+        return
+
+    # audio/video path
+    required = ['high', 'low', 'low_preview']
     conf = ec.preview_conf or {}
     intervals = conf.get('intervals') or [[0, int(conf.get('duration_ms', 30000))]]
     main_interval = intervals[0]
-    trim_value = None
     start_s = max(0, int(main_interval[0]) // 1000)
     dur_s = max(1, int((main_interval[1] - main_interval[0]) // 1000) or 30)
-    trim_value = f"{start_s},{dur_s}"
+    trim_value = f"{start_s}-{start_s + dur_s}"
 
     qualities = {
         'high': 'high',
@@ -133,81 +240,128 @@ async def _convert_content(ec: EncryptedContent, input_host_path: str):
         'low_preview': 'low',
     }
 
+    all_success = True
+    errors: List[str] = []
+
     for opt in required:
+        derivative_kind = f"decrypted_{opt if opt != 'low_preview' else 'preview'}"
+        derivative_id: Optional[int] = None
         try:
-            # Mark derivative processing
             async with db_session() as session:
                 cd = ContentDerivative(
                     content_id=ec.id,
-                    kind=f"decrypted_{opt if opt != 'low_preview' else 'preview'}",
+                    kind=derivative_kind,
                     interval_start_ms=main_interval[0] if opt == 'low_preview' else None,
                     interval_end_ms=main_interval[1] if opt == 'low_preview' else None,
                     local_path="",
                     status='processing',
                 )
                 session.add(cd)
+                await session.flush()
+                derivative_id = cd.id
                 await session.commit()
 
             out_path, ffprobe = await _run_media_converter(
-                input_host_path=input_host_path,
+                staging=staging,
                 input_ext=input_ext,
                 quality=qualities[opt],
                 trim_value=trim_value if opt == 'low_preview' else None,
                 is_audio=is_audio,
             )
 
-            # Save into store and StoredContent
             file_hash, size_bytes = await _save_derivative(out_path, os.path.basename(out_path))
 
             async with db_session() as session:
-                sc = StoredContent(
-                    type="local/content_bin",
-                    hash=file_hash,
-                    user_id=None,
-                    filename=os.path.basename(out_path),
-                    meta={'encrypted_cid': ec.encrypted_cid, 'kind': opt, 'ffprobe_meta': ffprobe},
-                    created=datetime.utcnow(),
-                )
-                session.add(sc)
-                await session.flush()
+                sc = (await session.execute(select(StoredContent).where(StoredContent.hash == file_hash))).scalars().first()
+                meta_payload = {'encrypted_cid': ec.encrypted_cid, 'kind': opt, 'ffprobe_meta': ffprobe}
+                if sc:
+                    sc.type = sc.type or "local/content_bin"
+                    sc.filename = os.path.basename(out_path)
+                    sc.meta = meta_payload
+                    sc.updated = datetime.utcnow()
+                else:
+                    sc = StoredContent(
+                        type="local/content_bin",
+                        hash=file_hash,
+                        user_id=None,
+                        filename=os.path.basename(out_path),
+                        meta=meta_payload,
+                        created=datetime.utcnow(),
+                    )
+                    session.add(sc)
+                    await session.flush()
 
-                # Update derivative record
-                cd = (await session.execute(select(ContentDerivative).where(
-                    ContentDerivative.content_id == ec.id,
-                    ContentDerivative.kind == (f"decrypted_{opt if opt != 'low_preview' else 'preview'}"),
-                    ContentDerivative.status == 'processing'
-                ))).scalars().first()
+                encrypted_sc = (await session.execute(select(StoredContent).where(StoredContent.hash == encrypted_hash_b58))).scalars().first()
+                if encrypted_sc:
+                    meta = dict(encrypted_sc.meta or {})
+                    converted = dict(meta.get('converted_content') or {})
+                    converted[opt] = file_hash
+                    meta['converted_content'] = converted
+                    encrypted_sc.meta = meta
+                    if opt == 'high':
+                        encrypted_sc.decrypted_content_id = sc.id
+                    encrypted_sc.updated = datetime.utcnow()
+
+                cd = await session.get(ContentDerivative, derivative_id) if derivative_id else None
                 if cd:
                     cd.local_path = os.path.join(UPLOADS_DIR, file_hash)
                     cd.size_bytes = size_bytes
-                    cd.content_type = ('audio/mpeg' if is_audio else 'video/mp4') if opt != 'high' else ec.content_type
+                    if is_audio:
+                        cd.content_type = 'audio/flac' if opt == 'high' else 'audio/mpeg'
+                    else:
+                        cd.content_type = ec.content_type if opt == 'high' else 'video/mp4'
                     cd.status = 'ready'
+                    cd.error = None
                 await session.commit()
 
+            output_parent = Path(out_path).parent
+            shutil.rmtree(output_parent, ignore_errors=True)
             make_log('convert_v3', f"Converted {ec.encrypted_cid} opt={opt} -> {file_hash}")
         except Exception as e:
             make_log('convert_v3', f"Convert error {ec.encrypted_cid} opt={opt}: {e}", level='error')
+            all_success = False
+            errors.append(_short_error(e))
             async with db_session() as session:
-                cd = ContentDerivative(
-                    content_id=ec.id,
-                    kind=f"decrypted_{opt if opt != 'low_preview' else 'preview'}",
-                    status='failed',
-                    error=str(e),
-                    local_path="",
-                )
-                session.add(cd)
+                cd = await session.get(ContentDerivative, derivative_id) if derivative_id else None
+                if cd:
+                    cd.status = 'failed'
+                    cd.error = _short_error(e)
+                else:
+                    session.add(ContentDerivative(
+                        content_id=ec.id,
+                        kind=derivative_kind,
+                        status='failed',
+                        error=_short_error(e),
+                        local_path="",
+                    ))
                 await session.commit()
 
+    await _update_upload_session(ec, all_success, errors)
 
-async def _pick_pending(limit: int) -> List[Tuple[EncryptedContent, str]]:
+
+async def _pick_pending(limit: int) -> List[Tuple[EncryptedContent, PlainStaging]]:
     async with db_session() as session:
         # Find A/V contents with preview_enabled and no ready low/low_preview derivatives yet
         ecs = (await session.execute(select(EncryptedContent).where(
             EncryptedContent.preview_enabled == True
         ).order_by(EncryptedContent.created_at.desc()))).scalars().all()
 
-        picked: List[Tuple[EncryptedContent, str]] = []
+        picked: List[Tuple[EncryptedContent, PlainStaging]] = []
         for ec in ecs:
+            try:
+                cid_obj, cid_err = resolve_content(ec.encrypted_cid)
+                if cid_err:
+                    make_log('convert_v3', f"Skip {ec.encrypted_cid}: resolve error {cid_err}", level='debug')
+                    continue
+                encrypted_hash_b58 = cid_obj.content_hash_b58
+            except Exception as exc:
+                make_log('convert_v3', f"Skip {ec.encrypted_cid}: resolve exception {exc}", level='warning')
+                continue
+
+            sc = (await session.execute(select(StoredContent).where(StoredContent.hash == encrypted_hash_b58))).scalars().first()
+            if not sc or sc.onchain_index is None:
+                continue
+
             # Check if derivatives already ready
             rows = (await session.execute(select(ContentDerivative).where(ContentDerivative.content_id == ec.id))).scalars().all()
             kinds_ready = {r.kind for r in rows if r.status == 'ready'}
@@ -215,11 +369,11 @@ async def _pick_pending(limit: int) -> List[Tuple[EncryptedContent, str]]:
             if required.issubset(kinds_ready):
                 continue
             # Always decrypt from IPFS using local or remote key
-            storage_path: Optional[str] = None
+            staging: Optional[PlainStaging] = None
             ck = (await session.execute(select(ContentKey).where(ContentKey.content_id == ec.id))).scalars().first()
             if ck:
-                storage_path = await stage_plain_from_ipfs(ec, ck.key_ciphertext_b64)
-            if not storage_path:
+                staging = await stage_plain_from_ipfs(ec, ck.key_ciphertext_b64)
+            if not staging:
                 peers = (await session.execute(select(KnownNode))).scalars().all()
                 for peer in peers:
                     base_url = f"http://{peer.ip}:{peer.port}"
@@ -240,12 +394,12 @@ async def _pick_pending(limit: int) -> List[Tuple[EncryptedContent, str]]:
                     )
                     session.add(session_ck)
                     await session.commit()
-                    storage_path = await stage_plain_from_ipfs(ec, dek_b64)
-                    if storage_path:
+                    staging = await stage_plain_from_ipfs(ec, dek_b64)
+                    if staging:
                         break
-            if not storage_path or not os.path.exists(storage_path):
+            if not staging or not os.path.exists(staging.container_path):
                 continue
-            picked.append((ec, storage_path))
+            picked.append((ec, staging))
             if len(picked) >= limit:
                 break
         return picked
@@ -254,14 +408,14 @@ async def _pick_pending(limit: int) -> List[Tuple[EncryptedContent, str]]:
 async def worker_loop():
     sem = asyncio.Semaphore(CONCURRENCY)
 
-    async def _run_one(ec: EncryptedContent, input_path: str):
+    async def _run_one(ec: EncryptedContent, staging: PlainStaging):
         async with sem:
             try:
-                await _convert_content(ec, input_path)
+                await _convert_content(ec, staging)
                 # After successful conversion, attempt to remove staging file to avoid duplicates
                 try:
-                    if input_path and input_path.startswith("/data/") and os.path.exists(input_path):
-                        os.remove(input_path)
+                    if staging and staging.container_path and os.path.exists(staging.container_path):
+                        os.remove(staging.container_path)
                 except Exception:
                     pass
             except Exception as e:
@@ -273,7 +427,7 @@ async def worker_loop():
             if not batch:
                 await asyncio.sleep(3)
                 continue
-            tasks = [asyncio.create_task(_run_one(ec, path)) for (ec, path) in batch]
+            tasks = [asyncio.create_task(_run_one(ec, staging)) for (ec, staging) in batch]
             await asyncio.gather(*tasks)
         except Exception as e:
             make_log('convert_v3', f"loop error: {e}", level='error')
@@ -285,15 +439,20 @@ async def main_fn(memory):
     await worker_loop()
 
 
-async def stage_plain_from_ipfs(ec: EncryptedContent, dek_wrapped: str) -> Optional[str]:
-    """Download encrypted ENCF stream from IPFS and decrypt on the fly into a temp file."""
-    import tempfile
+async def stage_plain_from_ipfs(ec: EncryptedContent, dek_wrapped: str) -> Optional[PlainStaging]:
+    """Download encrypted ENCF stream from IPFS and decrypt on the fly into shared staging."""
+    os.makedirs(UPLOADS_PATH / STAGING_SUBDIR, exist_ok=True)
     try:
         dek = unwrap_dek(dek_wrapped)
     except KeyWrapError as exc:
         make_log('convert_v3', f"unwrap failed for {ec.encrypted_cid}: {exc}", level='error')
         return None
-    tmp = tempfile.NamedTemporaryFile(prefix=f"dec_{ec.encrypted_cid[:8]}_", delete=False)
+
+    tmp = tempfile.NamedTemporaryFile(
+        prefix=f"dec_{ec.encrypted_cid[:8]}_",
+        dir=UPLOADS_PATH / STAGING_SUBDIR,
+        delete=False,
+    )
     tmp_path = tmp.name
     tmp.close()
     try:
@@ -301,7 +460,8 @@ async def stage_plain_from_ipfs(ec: EncryptedContent, dek_wrapped: str) -> Optio
             async for ch in cat_stream(ec.encrypted_cid):
                 yield ch
         await decrypt_encf_auto(_aiter(), dek, tmp_path)
-        return tmp_path
+        host_path = _container_to_host(tmp_path)
+        return PlainStaging(container_path=tmp_path, host_path=host_path)
     except Exception as e:
         make_log('convert_v3', f"decrypt from ipfs failed: {e}", level='error')
         try:
@@ -312,3 +472,8 @@ async def stage_plain_from_ipfs(ec: EncryptedContent, dek_wrapped: str) -> Optio
 
 
     
+def _short_error(message: str, limit: int = ERROR_TRUNCATE_LIMIT) -> str:
+    if not message:
+        return message
+    message = str(message)
+    return message if len(message) <= limit else message[: limit - 3] + '...'

app/core/background/derivative_cache_janitor.py

@@ -25,14 +25,14 @@ async def _evict_over_ttl(now: datetime) -> int:
     removed = 0
     # Pull TTL from ServiceConfig each time
     async with db_session() as session:
-        ttl_days = await ServiceConfig(session).get('DERIVATIVE_CACHE_TTL_DAYS', ENV_TTL_DAYS)
-    if int(ttl_days) <= 0:
+        ttl_days = int(await ServiceConfig(session).get('DERIVATIVE_CACHE_TTL_DAYS', ENV_TTL_DAYS))
+    if ttl_days <= 0:
         return 0
     async with db_session() as session:
         rows = (await session.execute(select(ContentDerivative).where(ContentDerivative.status == 'ready'))).scalars().all()
         for r in rows:
             la = r.last_access_at or r.created_at
-            if la and (now - la) > timedelta(days=TTL_DAYS):
+            if la and (now - la) > timedelta(days=ttl_days):
                 try:
                     if r.local_path and os.path.exists(r.local_path):
                         os.remove(r.local_path)
@@ -80,7 +80,11 @@ async def _evict_to_fit():
 
 
 async def main_fn(memory):
-    make_log('derivative_janitor', f"Started (MAX_GB={MAX_GB}, TTL_DAYS={TTL_DAYS})", level='info')
+    async with db_session() as session:
+        cfg = ServiceConfig(session)
+        runtime_max_gb = float(await cfg.get('DERIVATIVE_CACHE_MAX_GB', ENV_MAX_GB))
+        runtime_ttl_days = int(await cfg.get('DERIVATIVE_CACHE_TTL_DAYS', ENV_TTL_DAYS))
+    make_log('derivative_janitor', f"Started (MAX_GB={runtime_max_gb}, TTL_DAYS={runtime_ttl_days})", level='info')
     while True:
         try:
             now = datetime.utcnow()

app/core/background/index_scout_v3.py

@@ -1,5 +1,6 @@
 import asyncio
-from typing import List
+import os
+from typing import List, Optional
 
 import httpx
 import random
@@ -18,7 +19,7 @@ ENV_PIN_CONCURRENCY = int(os.getenv('SYNC_MAX_CONCURRENT_PINS', '4'))
 ENV_DISK_WATERMARK_PCT = int(os.getenv('SYNC_DISK_LOW_WATERMARK_PCT', '90'))
 
 
-async def fetch_index(base_url: str, etag: str | None, since: str | None) -> tuple[List[dict], str | None]:
+async def fetch_index(base_url: str, etag: Optional[str], since: Optional[str]) -> tuple[List[dict], Optional[str]]:
     try:
         headers = {}
         params = {}

app/core/background/indexer_service.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 from base64 import b64decode
 from datetime import datetime
 
@@ -6,7 +7,7 @@ from base58 import b58encode
 from sqlalchemy import String, and_, desc, cast
 from tonsdk.boc import Cell
 from tonsdk.utils import Address
-from app.core._config import CLIENT_TELEGRAM_BOT_USERNAME
+from app.core._config import CLIENT_TELEGRAM_BOT_USERNAME, PROJECT_HOST
 from app.core._blockchain.ton.platform import platform
 from app.core._blockchain.ton.toncenter import toncenter
 from app.core._utils.send_status import send_status
@@ -18,6 +19,10 @@ from app.core._utils.resolve_content import resolve_content
 from app.core.models.wallet_connection import WalletConnection
 from app.core._keyboards import get_inline_keyboard
 from app.core.models._telegram import Wrapped_CBotChat
+from app.core._utils.share_links import build_content_links
+
+
+MIN_ONCHAIN_INDEX = int(os.getenv("MIN_ONCHAIN_INDEX", "8"))
 from sqlalchemy import select, and_, desc
 from app.core.storage import db_session
 import os
@@ -110,11 +115,15 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
         )).scalars().first()
         last_known_index = last_known_index_.onchain_index if last_known_index_ else 0
         last_known_index = max(last_known_index, 0)
+        if last_known_index < (MIN_ONCHAIN_INDEX - 1):
+            make_log(
+                "Indexer",
+                f"Adjusting last_known_index from {last_known_index} to {MIN_ONCHAIN_INDEX - 1} (MIN_ONCHAIN_INDEX)",
+                level="debug"
+            )
+            last_known_index = MIN_ONCHAIN_INDEX - 1
         make_log("Indexer", f"Last known index: {last_known_index}", level="debug")
-        if last_known_index_:
-            next_item_index = last_known_index + 1
-        else:
-            next_item_index = 0
+        next_item_index = last_known_index + 1
 
         resolve_item_result = await toncenter.run_get_method(platform.address.to_string(1, 1, 1), 'get_nft_address_by_index', [['num', next_item_index]])
         make_log("Indexer", f"Resolve item result: {resolve_item_result}", level="debug")
@@ -141,6 +150,13 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
 
         assert item_get_data_result['stack'][2][0] == 'num', "Item index is not a number"
         item_index = int(item_get_data_result['stack'][2][1], 16)
+        if item_index < MIN_ONCHAIN_INDEX:
+            make_log(
+                "Indexer",
+                f"Skip on-chain item {item_index}: below MIN_ONCHAIN_INDEX={MIN_ONCHAIN_INDEX}",
+                level="info"
+            )
+            return platform_found, seqno
         assert item_index == next_item_index, "Item index mismatch"
 
         item_platform_address = Cell.one_from_boc(b64decode(item_get_data_result['stack'][3][1]['bytes'])).begin_parse().read_msg_addr()
@@ -222,17 +238,32 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
 
                     if user:
                         user_uploader_wrapper = Wrapped_CBotChat(memory._telegram_bot, chat_id=user.telegram_id, user=user, db_session=session)
+                        ref_id = (user.meta or {}).get('ref_id')
+                        if not ref_id:
+                            ref_id = user.ensure_ref_id()
+                            await session.commit()
+
+                        _, startapp_url, web_url = build_content_links(
+                            encrypted_stored_content.cid.serialize_v2(),
+                            ref_id,
+                            project_host=PROJECT_HOST,
+                            bot_username=CLIENT_TELEGRAM_BOT_USERNAME
+                        )
+
+                        message_text = user.translated('p_contentWasIndexed').format(
+                            item_address=item_address.to_string(1, 1, 1),
+                            item_index=item_index,
+                        )
+                        message_text += f"\n\n<a href=\"{web_url}\">🔗 Открыть контент</a>"
+
                         await user_uploader_wrapper.send_message(
-                            user.translated('p_contentWasIndexed').format(
-                                item_address=item_address.to_string(1, 1, 1),
-                                item_index=item_index,
-                            ),
+                            message_text,
                             message_type='notification',
                             reply_markup=get_inline_keyboard([
                                 [{
                                     'text': user.translated('viewTrackAsClient_button'),
-                                    'url': f"https://t.me/{CLIENT_TELEGRAM_BOT_USERNAME}?start=C{encrypted_stored_content.cid.serialize_v2()}"
-                                }],
+                                    'url': startapp_url
+                                }]
                             ])
                         )
 
@@ -263,6 +294,7 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
                     **encrypted_stored_content.meta,
                     **item_metadata_packed
                 }
+                encrypted_stored_content.content_id = item_content_cid_str
 
                 await session.commit()
                 return platform_found, seqno
@@ -283,6 +315,7 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
             encrypted=True,
             decrypted_content_id=None,
             key_id=None,
+            content_id=item_content_cid_str,
             updated=datetime.now()
         )
         session.add(onchain_stored_content)

app/core/background/license_service.py

@@ -100,6 +100,7 @@ async def license_index_loop(memory, platform_found: bool, seqno: int) -> [bool,
         process_content = (await session.execute(select(UserContent).where(
             and_(
                 UserContent.type.startswith('nft/'),
+                UserContent.type != 'nft/ignored',
                 UserContent.updated < (datetime.now() - timedelta(minutes=60)),
             )
         ).order_by(UserContent.updated.asc()))).scalars().first()