diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 808b8c5..f6993a2 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -14,6 +14,7 @@ env: REGISTRY: ghcr.io IMAGE_API: ${{ github.repository }}/api IMAGE_WEB: ${{ github.repository }}/web + IMAGE_MONITORING: ${{ github.repository }}/monitoring jobs: build-and-push: @@ -82,6 +83,31 @@ jobs: cache-from: type=gha,scope=web cache-to: type=gha,mode=max,scope=web + # ── Monitoring ─────────────────────────────────────────────────────────── + + - name: Extract metadata for Monitoring image + id: meta-monitoring + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_MONITORING }} + tags: | + type=ref,event=branch + type=sha,prefix={{branch}}-,format=short + + - name: Build and push Monitoring image + uses: docker/build-push-action@v7 + with: + context: . + file: apps/monitoring/Dockerfile + push: true + tags: ${{ steps.meta-monitoring.outputs.tags }} + labels: ${{ steps.meta-monitoring.outputs.labels }} + build-args: | + NEXT_PUBLIC_API_BASE_URL= + MONITORING_BASE_PATH=/ops + cache-from: type=gha,scope=monitoring + cache-to: type=gha,mode=max,scope=monitoring + # ── Deploy（按需取消注释）────────────────────────────────────────────────── # # deploy: diff --git a/apps/api/app/services/source_service.py b/apps/api/app/services/source_service.py index b403d86..6f24dd6 100644 --- a/apps/api/app/services/source_service.py +++ b/apps/api/app/services/source_service.py @@ -540,9 +540,12 @@ async def rechunk_source( source.status = "pending" await self.db.flush() + trace_id, run = await self._create_source_ingest_run(source, origin="rechunk") self._enqueue_ingestion( source.id, + trace_id=trace_id, + run_id=run.id, job_kind="rechunk", job_label=f"重建索引：{source.title or source.id}", chunk_size=size, @@ -628,16 +631,13 @@ async def upload_global_source(self, filename: str | None, content: bytes) -> So nb = await self._get_or_create_global_notebook() ext = os.path.splitext(filename or "")[1].lower() - type_map = {".pdf": "pdf", ".md": "md", ".txt": "md"} + type_map = {".pdf": "pdf", ".md": "md", ".txt": "md", ".docx": "doc"} source_type = type_map.get(ext, "md") file_id = str(uuid.uuid4()) storage_key = f"global/{nb.id}/{file_id}{ext}" content_type = self._guess_content_type(filename or "") - from app.providers.storage import storage as get_storage - await get_storage().upload(storage_key, content, content_type) - source = Source( notebook_id=nb.id, title=filename, @@ -649,9 +649,30 @@ async def upload_global_source(self, filename: str | None, content: bytes) -> So self.db.add(source) await self.db.flush() await self.db.refresh(source) + trace_id, run = await self._create_source_ingest_run(source, origin="global_upload") + from app.providers.storage import storage as get_storage + upload_started_at = datetime.now(UTC) + await get_storage().upload(storage_key, content, content_type) + await record_completed_span( + self.db, + "source_ingest.upload", + run=run, + component="api", + span_kind="phase", + status="succeeded", + metadata={ + "content_type": content_type, + "filename": filename, + "byte_length": len(content), + }, + started_at=upload_started_at, + finished_at=datetime.now(UTC), + ) self._enqueue_ingestion( source.id, + trace_id=trace_id, + run_id=run.id, job_kind="import", job_label=f"索引资料：{filename or '未命名文件'}", ) @@ -672,9 +693,23 @@ async def import_global_source_url(self, url: str, title: str | None) -> Source: self.db.add(source) await self.db.flush() await self.db.refresh(source) + trace_id, run = await self._create_source_ingest_run( + source, origin="global_url_import" + ) + await record_instant_span( + self.db, + "source_ingest.upload", + run=run, + component="api", + span_kind="phase", + status="succeeded", + metadata={"mode": "url", "url": url, "skipped": True}, + ) self._enqueue_ingestion( source.id, + trace_id=trace_id, + run_id=run.id, job_kind="import", job_label=f"索引网页：{title or url}", ) diff --git a/apps/api/tests/unit/test_source_service_dispatch.py b/apps/api/tests/unit/test_source_service_dispatch.py index 2000026..683dd13 100644 --- a/apps/api/tests/unit/test_source_service_dispatch.py +++ b/apps/api/tests/unit/test_source_service_dispatch.py @@ -104,6 +104,163 @@ async def flush() -> None: db.commit.assert_awaited_once() +@pytest.mark.asyncio +async def test_upload_global_source_dispatches_ingestion_after_commit(monkeypatch) -> None: + added: list[object] = [] + + async def flush() -> None: + for obj in added: + if isinstance(obj, Source) and getattr(obj, "id", None) is None: + setattr(obj, "id", uuid.uuid4()) + if isinstance(obj, ObservabilityRun) and getattr(obj, "id", None) is None: + setattr(obj, "id", uuid.uuid4()) + + db = SimpleNamespace( + add=Mock(side_effect=added.append), + flush=AsyncMock(side_effect=flush), + refresh=AsyncMock(return_value=None), + commit=AsyncMock(return_value=None), + sync_session=SimpleNamespace(info={}), + ) + service = SourceService(db, uuid.uuid4()) + service._get_or_create_global_notebook = AsyncMock( # type: ignore[method-assign] + return_value=SimpleNamespace(id=uuid.uuid4()) + ) + + storage = SimpleNamespace(upload=AsyncMock(return_value=None)) + monkeypatch.setattr("app.providers.storage.storage", lambda: storage) + + delayed: list[tuple[str, dict[str, object]]] = [] + monkeypatch.setattr( + "app.workers.tasks.ingest_source.delay", + lambda source_id, **kwargs: delayed.append((source_id, kwargs)), + ) + + source = await service.upload_global_source("global.docx", b"doc-bytes") + + assert source.status == "pending" + assert source.type == "doc" + assert delayed == [] + + callbacks = db.sync_session.info.get("_after_commit_callbacks", []) + assert len(callbacks) == 1 + callbacks[0]() + + assert delayed[0][0] == str(source.id) + assert delayed[0][1]["trace_id"] + assert delayed[0][1]["run_id"] + storage.upload.assert_awaited_once() + db.commit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_import_global_source_url_dispatches_ingestion_after_commit( + monkeypatch, +) -> None: + added: list[object] = [] + + async def flush() -> None: + for obj in added: + if isinstance(obj, Source) and getattr(obj, "id", None) is None: + setattr(obj, "id", uuid.uuid4()) + if isinstance(obj, ObservabilityRun) and getattr(obj, "id", None) is None: + setattr(obj, "id", uuid.uuid4()) + + db = SimpleNamespace( + add=Mock(side_effect=added.append), + flush=AsyncMock(side_effect=flush), + refresh=AsyncMock(return_value=None), + commit=AsyncMock(return_value=None), + sync_session=SimpleNamespace(info={}), + ) + service = SourceService(db, uuid.uuid4()) + service._get_or_create_global_notebook = AsyncMock( # type: ignore[method-assign] + return_value=SimpleNamespace(id=uuid.uuid4()) + ) + + delayed: list[tuple[str, dict[str, object]]] = [] + monkeypatch.setattr( + "app.workers.tasks.ingest_source.delay", + lambda source_id, **kwargs: delayed.append((source_id, kwargs)), + ) + + source = await service.import_global_source_url( + "https://example.com/global", + "Global Example", + ) + + assert source.status == "pending" + assert delayed == [] + + callbacks = db.sync_session.info.get("_after_commit_callbacks", []) + assert len(callbacks) == 1 + callbacks[0]() + + assert delayed[0][0] == str(source.id) + assert delayed[0][1]["trace_id"] + assert delayed[0][1]["run_id"] + db.commit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_rechunk_source_dispatches_ingestion_after_commit_with_trace_context( + monkeypatch, +) -> None: + added: list[object] = [] + + async def flush() -> None: + for obj in added: + if isinstance(obj, ObservabilityRun) and getattr(obj, "id", None) is None: + setattr(obj, "id", uuid.uuid4()) + + db = SimpleNamespace( + add=Mock(side_effect=added.append), + flush=AsyncMock(side_effect=flush), + commit=AsyncMock(return_value=None), + sync_session=SimpleNamespace(info={}), + ) + service = SourceService(db, uuid.uuid4()) + source = SimpleNamespace( + id=uuid.uuid4(), + notebook_id=uuid.uuid4(), + title="demo.pdf", + type="pdf", + url=None, + status="indexed", + ) + monkeypatch.setattr(service, "_get_owned_source", AsyncMock(return_value=source)) + + apply_async_calls: list[dict[str, object]] = [] + monkeypatch.setattr( + "app.workers.tasks.ingest_source.apply_async", + lambda *args, **kwargs: apply_async_calls.append( + {"args": list(args), "kwargs": kwargs} + ), + ) + + chunk_size, chunk_overlap = await service.rechunk_source( + source.id, + strategy="standard", + ) + + assert source.status == "pending" + assert chunk_size == 512 + assert chunk_overlap == 64 + assert apply_async_calls == [] + + callbacks = db.sync_session.info.get("_after_commit_callbacks", []) + assert len(callbacks) == 1 + callbacks[0]() + + dispatched = apply_async_calls[0] + assert dispatched["kwargs"]["args"] == [str(source.id)] + assert dispatched["kwargs"]["kwargs"]["trace_id"] + assert dispatched["kwargs"]["kwargs"]["run_id"] + assert dispatched["kwargs"]["kwargs"]["chunk_size"] == chunk_size + assert dispatched["kwargs"]["kwargs"]["chunk_overlap"] == chunk_overlap + db.commit.assert_awaited_once() + + @pytest.mark.asyncio async def test_import_global_source_path_reads_file_and_delegates( tmp_path, diff --git a/packages/cli/src/commands/prod.js b/packages/cli/src/commands/prod.js index 1bb2996..d244b99 100644 --- a/packages/cli/src/commands/prod.js +++ b/packages/cli/src/commands/prod.js @@ -1,10 +1,43 @@ import fs from 'fs'; import path from 'path'; import ora from 'ora'; -import { section, log, warn, info } from '../utils/ui.js'; -import { exec } from '../utils/proc.js'; +import { section, warn, info } from '../utils/ui.js'; +import { exec, execQ } from '../utils/proc.js'; import { ROOT_DIR } from '../utils/paths.js'; +export const PROD_UPDATE_PULL_SERVICES = ['api', 'web', 'monitoring']; + +export function parseTrackedFilesFromGitStatus(statusOutput) { + return statusOutput + .split('\n') + .map((line) => line.trimEnd()) + .filter(Boolean) + .filter((line) => !line.startsWith('?? ')) + .map((line) => line.slice(3)); +} + +export function buildProdUpdateDirtyWorktreeGuidance(files) { + const visibleFiles = files.slice(0, 10); + const lines = [ + '检测到本地未提交的 Git 改动，已停止更新以避免覆盖这些文件：', + ...visibleFiles.map((file) => `- ${file}`), + ]; + + if (files.length > visibleFiles.length) { + lines.push(`- 以及另外 ${files.length - visibleFiles.length} 个文件`); + } + + lines.push('如果这些改动不需要保留，请先执行：git restore <文件>'); + lines.push('如果这些改动需要暂存，请先执行：git stash push --include-untracked'); + lines.push('清理完成后，再重新运行 lyra update。'); + + return lines; +} + +export function getProdUpdatePullCommand() { + return `docker compose -f docker-compose.prod.yml pull ${PROD_UPDATE_PULL_SERVICES.join(' ')}`; +} + export async function startProd() { section('生产模式启动（ghcr.io 云端镜像）'); @@ -39,10 +72,19 @@ export async function updateProd() { section('一键更新（git pull + 拉新镜像 + 重启）'); process.chdir(ROOT_DIR); + const dirtyTrackedFiles = parseTrackedFilesFromGitStatus(execQ('git status --porcelain')); + if (dirtyTrackedFiles.length > 0) { + warn('检测到本地改动，已取消本次更新。'); + for (const line of buildProdUpdateDirtyWorktreeGuidance(dirtyTrackedFiles)) { + info(line); + } + process.exit(1); + } + const spinner = ora('更新到最新版本...').start(); try { - exec('git pull', { shell: true }); - exec('docker compose -f docker-compose.prod.yml pull web api worker', { shell: true }); + exec('git pull --ff-only', { shell: true }); + exec(getProdUpdatePullCommand(), { shell: true }); exec('docker compose -f docker-compose.prod.yml up -d', { shell: true }); exec('docker image prune -f', { shell: true }); spinner.succeed('更新完成'); diff --git a/packages/cli/src/commands/prod.test.js b/packages/cli/src/commands/prod.test.js new file mode 100644 index 0000000..0c59928 --- /dev/null +++ b/packages/cli/src/commands/prod.test.js @@ -0,0 +1,44 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { + buildProdUpdateDirtyWorktreeGuidance, + getProdUpdatePullCommand, + parseTrackedFilesFromGitStatus, + PROD_UPDATE_PULL_SERVICES, +} from './prod.js'; + +test('parseTrackedFilesFromGitStatus ignores untracked files', () => { + const status = [ + ' M pnpm-lock.yaml', + 'A packages/cli/src/commands/prod.test.js', + '?? apps/api/tmp/debug.log', + ].join('\n'); + + assert.deepEqual(parseTrackedFilesFromGitStatus(status), [ + 'pnpm-lock.yaml', + 'packages/cli/src/commands/prod.test.js', + ]); +}); + +test('buildProdUpdateDirtyWorktreeGuidance lists files and recovery steps', () => { + const lines = buildProdUpdateDirtyWorktreeGuidance([ + 'pnpm-lock.yaml', + 'packages/cli/src/commands/prod.js', + ]); + + assert.equal(lines[0], '检测到本地未提交的 Git 改动，已停止更新以避免覆盖这些文件：'); + assert.equal(lines[1], '- pnpm-lock.yaml'); + assert.equal(lines[2], '- packages/cli/src/commands/prod.js'); + assert.ok(lines.includes('如果这些改动不需要保留，请先执行：git restore <文件>')); + assert.ok(lines.includes('如果这些改动需要暂存，请先执行：git stash push --include-untracked')); + assert.equal(lines.at(-1), '清理完成后，再重新运行 lyra update。'); +}); + +test('getProdUpdatePullCommand includes monitoring image', () => { + assert.deepEqual(PROD_UPDATE_PULL_SERVICES, ['api', 'web', 'monitoring']); + assert.equal( + getProdUpdatePullCommand(), + 'docker compose -f docker-compose.prod.yml pull api web monitoring' + ); +});