diff --git a/app/business/extension/AGENTS.md b/app/business/extension/AGENTS.md
index c623e84..f7eac41 100644
--- a/app/business/extension/AGENTS.md
+++ b/app/business/extension/AGENTS.md
@@ -8,6 +8,10 @@
- 一个 canonical `extensions` row 表示 deployment 安装的 exact Release；`installed`、
`enabled[]`、`running` 不可混用。
- `ExtensionBase` 保持配置校验、`on_start`、`on_close` 与可逆 route/source/resolver publication。
+- `ExtensionBase.get_config()`、`get_state()` 读取当前数据库真值；config/state mutation 通过 Host
+ 提供的语义接口持久化。Extension 不接触 SQLModel relation，且 Host 不缓存 state 真值。
+- `config` 是用户声明的部署配置；`state` 是 Extension 管理的 deployment-wide 运行状态。
+ 并发与事务由 Core/PostgreSQL adapter 负责，不要求 Extension 自己实现 compare-and-set。
- Extension wheel 直接 import Core 模块；Host 只接受标准 `inkcre.core.extensions` entry point。
- Registry Simple URL 必须与配置的 Registry 同源且路径精确匹配 Project。
- runtime 只能把普通 wheel 安装到当前 Core interpreter/site-packages；禁止 `pip --target`、
@@ -21,6 +25,8 @@
- enable 先启动 runtime，再调用 atomic enabled RPC；返回 version 不一致时移除 peer 并停止旧 runtime。
- disable 先停止 runtime，再调用 RPC；RPC 失败时重启 exact prior runtime，durable intent 不变。
- cold restore 失败不得删除 `enabled[]`，并使 bootstrap/readiness 失败。
+- 公开 HTTP 入口必须由 Extension 显式声明 exact method/path；Host 仅在 runtime publication
+ 存活期间授权该 route 绕过 Peer JWT，撤销 runtime 时同步撤销授权。
## 权限和持久化
diff --git a/app/business/extension/__init__.py b/app/business/extension/__init__.py
index 81551ab..643aed3 100644
--- a/app/business/extension/__init__.py
+++ b/app/business/extension/__init__.py
@@ -1,9 +1,19 @@
-from .main import EXTENSION_HOST, ExtensionBase, ExtensionHost
-from .state import ExtensionState
+from .main import (
+ EXTENSION_HOST,
+ EmptyConfig,
+ EmptyState,
+ ExtensionBase,
+ ExtensionHost,
+ PublicHTTPRoute,
+)
+from .state import InstalledExtension
__all__ = [
"EXTENSION_HOST",
+ "EmptyConfig",
+ "EmptyState",
"ExtensionBase",
"ExtensionHost",
- "ExtensionState",
+ "InstalledExtension",
+ "PublicHTTPRoute",
]
diff --git a/app/business/extension/main.py b/app/business/extension/main.py
index f85e07e..2a14224 100644
--- a/app/business/extension/main.py
+++ b/app/business/extension/main.py
@@ -39,11 +39,13 @@
from .runtime import (
ExtensionPublication,
ExtensionPublicationSnapshot,
+ PublicHTTPRoute,
+ PublicHTTPRouteClaim,
ExtensionRuntimeClaim,
ExtensionRuntimeClaimConflictError,
ExtensionRuntimeRecord,
)
-from .state import ExtensionState, ExtensionStateStore, SQLExtensionStateStore
+from .state import ExtensionStore, InstalledExtension, SQLExtensionStore
LOGGER = get_logger().getChild(__name__)
@@ -52,6 +54,9 @@
class EmptyConfig(sqlmodel.SQLModel): ...
+class EmptyState(sqlmodel.SQLModel): ...
+
+
ConfigTV = typing.TypeVar("ConfigTV", bound=sqlmodel.SQLModel)
@@ -64,11 +69,13 @@ def __init_subclass__(
cls,
ext_id: str,
config_cls: type[ConfigTV],
+ state_cls: type[sqlmodel.SQLModel] = EmptyState,
**kwargs: typing.Any,
) -> None:
cls.__extid__ = ext_id
cls.__configcls__ = config_cls # pyrefly: ignore[no-access]
cls.__configschema__ = config_cls.model_json_schema()
+ cls.__statecls__ = state_cls
super().__init_subclass__(**kwargs)
@classmethod
@@ -98,10 +105,16 @@ def on_start(
setattr(cls, "config", validated_config)
router = fastapi.APIRouter(prefix=f"/{cls.__extid__}")
cls._register_apis(router)
+ registered_routes = tuple(router.routes)
app.include_router(router, tags=["extension", cls.__extid__])
cls._init_sources()
cls._init_resolvers()
publication = snapshot.finish()
+ publication.public_http_claim = PublicHTTPRouteClaim.acquire(
+ cls.__extid__,
+ cls.public_http_routes(),
+ registered_routes,
+ )
publication.activate_source_types()
extension.persist_config_schema(dict(cls.__configschema__))
except Exception:
@@ -128,8 +141,6 @@ async def on_close(cls) -> None:
)
if record is None:
raise ExtensionRuntimeError(f"Extension runtime {cls.__extid__} has no state")
- config = typing.cast(sqlmodel.SQLModel, getattr(cls, "config"))
- record.persist_config(config.model_dump())
LOGGER.info("Extension %s closed", cls.__extid__)
@classmethod
@@ -161,12 +172,81 @@ def _register_apis(cls, router: fastapi.APIRouter) -> None:
"""Register Extension-owned API endpoints."""
@classmethod
- def update_config(cls, new_config: dict[str, typing.Any] | ConfigTV) -> None:
+ def public_http_routes(cls) -> tuple[PublicHTTPRoute, ...]:
+ return ()
+
+ @classmethod
+ def _runtime_record(cls) -> ExtensionRuntimeRecord:
+ record = typing.cast(
+ ExtensionRuntimeRecord | None,
+ cls.__dict__.get("__runtime_record__"),
+ )
+ if record is None:
+ raise ExtensionRuntimeError(f"Extension runtime {cls.__extid__} has no state")
+ return record
+
+ @classmethod
+ def get_config(cls) -> ConfigTV:
+ validated = cls.__configcls__( # pyrefly: ignore[missing-attribute]
+ **cls._runtime_record().read_config()
+ )
+ setattr(cls, "config", validated)
+ return validated
+
+ @classmethod
+ def update_config(cls, new_config: dict[str, typing.Any] | ConfigTV) -> ConfigTV:
if isinstance(new_config, dict):
validated = cls.__configcls__(**new_config) # pyrefly: ignore[missing-attribute]
else:
validated = new_config
+ cls._runtime_record().persist_config(validated.model_dump(mode="json"))
setattr(cls, "config", validated)
+ return validated
+
+ @classmethod
+ def get_state(cls) -> sqlmodel.SQLModel:
+ return cls.__statecls__(**cls._runtime_record().read_state()) # pyrefly: ignore[missing-attribute]
+
+ @classmethod
+ def mutate_state(
+ cls,
+ transform: typing.Callable[[sqlmodel.SQLModel], sqlmodel.SQLModel],
+ ) -> sqlmodel.SQLModel:
+ state_cls = cls.__statecls__ # pyrefly: ignore[missing-attribute]
+
+ def mutate(raw: dict[str, typing.Any]) -> dict[str, typing.Any]:
+ current = state_cls(**raw)
+ updated = transform(current)
+ if not isinstance(updated, state_cls):
+ raise TypeError("Extension state transform returned the wrong model")
+ return updated.model_dump(mode="json")
+
+ return state_cls(**cls._runtime_record().mutate_state(mutate))
+
+ @classmethod
+ def mutate_config_and_state(
+ cls,
+ transform: typing.Callable[
+ [ConfigTV, sqlmodel.SQLModel], tuple[ConfigTV, sqlmodel.SQLModel]
+ ],
+ ) -> tuple[ConfigTV, sqlmodel.SQLModel]:
+ config_cls = cls.__configcls__ # pyrefly: ignore[missing-attribute]
+ state_cls = cls.__statecls__ # pyrefly: ignore[missing-attribute]
+
+ def mutate(
+ raw_config: dict[str, typing.Any],
+ raw_state: dict[str, typing.Any],
+ ) -> tuple[dict[str, typing.Any], dict[str, typing.Any]]:
+ config, state = transform(config_cls(**raw_config), state_cls(**raw_state))
+ if not isinstance(config, config_cls) or not isinstance(state, state_cls):
+ raise TypeError("Extension config/state transform returned the wrong models")
+ return config.model_dump(mode="json"), state.model_dump(mode="json")
+
+ raw_config, raw_state = cls._runtime_record().mutate_config_and_state(mutate)
+ config = config_cls(**raw_config)
+ state = state_cls(**raw_state)
+ setattr(cls, "config", config)
+ return config, state
@dataclass
@@ -186,11 +266,11 @@ class ExtensionHost:
def __init__(
self,
*,
- store: ExtensionStateStore | None = None,
+ store: ExtensionStore | None = None,
release_client: ReleaseResolver | None = None,
distribution_consumer: DistributionConsumer | None = None,
) -> None:
- self.store = store or SQLExtensionStateStore()
+ self.store = store or SQLExtensionStore()
self.release_client = release_client or RegistryReleaseClient(
settings.extension_registry_url,
settings.extension_registry_timeout_seconds,
@@ -203,10 +283,10 @@ def __init__(
self._loaded_versions: dict[str, str] = {}
self._runtime_lock = asyncio.Lock()
- def list(self) -> tuple[ExtensionState, ...]:
+ def list(self) -> tuple[InstalledExtension, ...]:
return self.store.list()
- def get(self, name: str) -> ExtensionState:
+ def get(self, name: str) -> InstalledExtension:
validate_coordinate(name)
state = self.store.get(name)
if state is None:
@@ -230,7 +310,7 @@ def _resolve(
association = require_python_association(release)
return release, association
- def install(self, name: str, version: str) -> ExtensionState:
+ def install(self, name: str, version: str) -> InstalledExtension:
validate_coordinate(name, version)
existing = self.store.get(name)
if existing is not None and existing.version == version:
@@ -253,7 +333,7 @@ def update_config(
self,
name: str,
config: dict[str, typing.Any],
- ) -> ExtensionState:
+ ) -> InstalledExtension:
state = self.get(name)
running = self.running.get(name)
if running is None:
@@ -263,11 +343,10 @@ def update_config(
getattr(running.extension_class, "__configcls__"),
)
validated = config_class(**config)
- updated = self.store.update_config(name, validated.model_dump())
running.extension_class.update_config(validated)
- return updated
+ return self.get(name)
- def _acquire(self, state: ExtensionState):
+ def _acquire(self, state: InstalledExtension):
release, association = self._resolve(
state.name,
state.version,
@@ -279,7 +358,7 @@ def _acquire(self, state: ExtensionState):
async def _start(
self,
app: fastapi.FastAPI,
- state: ExtensionState,
+ state: InstalledExtension,
) -> RunningExtension:
existing = self.running.get(state.name)
if existing is not None:
@@ -295,7 +374,7 @@ async def _start(
async def _start_acquired(
self,
app: fastapi.FastAPI,
- state: ExtensionState,
+ state: InstalledExtension,
association: PythonReleaseDescriptor,
acquired: AcquiredDistribution,
*,
@@ -317,6 +396,25 @@ async def _start_acquired(
def persist_config(config: dict[str, typing.Any]) -> None:
self.store.update_config(state.name, config)
+ def read_config() -> dict[str, typing.Any]:
+ return self.store.read_config(state.name)
+
+ def read_state() -> dict[str, typing.Any]:
+ return self.store.read_state(state.name)
+
+ def mutate_state(
+ transform: typing.Callable[[dict[str, typing.Any]], dict[str, typing.Any]],
+ ) -> dict[str, typing.Any]:
+ return self.store.mutate_state(state.name, transform)
+
+ def mutate_config_and_state(
+ transform: typing.Callable[
+ [dict[str, typing.Any], dict[str, typing.Any]],
+ tuple[dict[str, typing.Any], dict[str, typing.Any]],
+ ],
+ ) -> tuple[dict[str, typing.Any], dict[str, typing.Any]]:
+ return self.store.mutate_config_and_state(state.name, transform)
+
def stage_schema(schema: dict[str, typing.Any]) -> None:
schema_box["value"] = schema
@@ -329,7 +427,11 @@ def stage_schema(schema: dict[str, typing.Any]) -> None:
runtime_record = ExtensionRuntimeRecord(
extension_id=association.entry_point.name,
config=dict(state.config),
+ read_config=read_config,
persist_config=persist_config,
+ read_state=read_state,
+ mutate_state=mutate_state,
+ mutate_config_and_state=mutate_config_and_state,
persist_config_schema=stage_schema,
)
extension_class.on_start(
@@ -404,7 +506,7 @@ async def enable(
name: str,
*,
app: fastapi.FastAPI | None = None,
- ) -> ExtensionState:
+ ) -> InstalledExtension:
validate_coordinate(name)
runtime_app = app or self.fastapi_app
if runtime_app is None:
@@ -445,7 +547,7 @@ async def enable(
) from conflict
raise conflict
- async def disable(self, name: str) -> ExtensionState:
+ async def disable(self, name: str) -> InstalledExtension:
validate_coordinate(name)
peer_id = ClientManager.get_current_client_id()
async with self._runtime_lock:
@@ -519,8 +621,10 @@ async def close_running(self) -> None:
__all__ = [
"EXTENSION_HOST",
"EmptyConfig",
+ "EmptyState",
"ExtensionBase",
"ExtensionHost",
"ExtensionHostError",
- "ExtensionState",
+ "InstalledExtension",
+ "PublicHTTPRoute",
]
diff --git a/app/business/extension/runtime.py b/app/business/extension/runtime.py
index 8fe7de0..3c27b27 100644
--- a/app/business/extension/runtime.py
+++ b/app/business/extension/runtime.py
@@ -1,4 +1,4 @@
-"""Reversible publication primitives shared by legacy and Registry extensions."""
+"""Reversible publication primitives for the native Extension Host."""
from __future__ import annotations
@@ -18,6 +18,85 @@
from app.schemas.info_base.block import ResolverType
+@dataclass(frozen=True)
+class PublicHTTPRoute:
+ """One exact Extension route that intentionally bypasses Peer JWT auth."""
+
+ method: typing.Literal["GET", "POST"]
+ path: str
+
+ def __post_init__(self) -> None:
+ if (
+ not self.path.startswith("/")
+ or self.path == "/"
+ or "{" in self.path
+ or "}" in self.path
+ or "*" in self.path
+ or "?" in self.path
+ or "#" in self.path
+ ):
+ raise ValueError("Public Extension route must be an exact relative path")
+
+
+class PublicHTTPRouteClaim:
+ """Lock-protected process authority for exact public Extension routes."""
+
+ _lock = threading.Lock()
+ _owners: dict[tuple[str, str], object] = {}
+
+ def __init__(self, routes: frozenset[tuple[str, str]], token: object) -> None:
+ self.routes = routes
+ self._token = token
+ self._released = False
+
+ @classmethod
+ def acquire(
+ cls,
+ extension_id: str,
+ declarations: tuple[PublicHTTPRoute, ...],
+ published_routes: tuple[typing.Any, ...],
+ ) -> PublicHTTPRouteClaim | None:
+ if not declarations:
+ return None
+ available = {
+ (method, route.path)
+ for route in published_routes
+ for method in (getattr(route, "methods", None) or ())
+ if isinstance(getattr(route, "path", None), str)
+ }
+ absolute = frozenset(
+ (declaration.method, f"/{extension_id}{declaration.path}")
+ for declaration in declarations
+ )
+ missing = absolute - available
+ if missing:
+ raise ValueError(f"Public Extension routes were not published: {sorted(missing)}")
+ token = object()
+ with cls._lock:
+ conflicts = absolute & cls._owners.keys()
+ if conflicts:
+ raise ExtensionRuntimeClaimConflictError(
+ f"Public Extension route already claimed: {sorted(conflicts)}"
+ )
+ for route in absolute:
+ cls._owners[route] = token
+ return cls(absolute, token)
+
+ @classmethod
+ def permits(cls, method: str, path: str) -> bool:
+ with cls._lock:
+ return (method.upper(), path) in cls._owners
+
+ def release(self) -> None:
+ if self._released:
+ return
+ with self._lock:
+ for route in self.routes:
+ if self._owners.get(route) is self._token:
+ self._owners.pop(route)
+ self._released = True
+
+
class ExtensionRuntimeClaimConflictError(RuntimeError):
"""Raised when another manager already owns an Extension runtime ID."""
@@ -60,7 +139,22 @@ class ExtensionRuntimeRecord:
extension_id: str
config: dict[str, typing.Any]
+ read_config: Callable[[], dict[str, typing.Any]]
persist_config: Callable[[dict[str, typing.Any]], None]
+ read_state: Callable[[], dict[str, typing.Any]]
+ mutate_state: Callable[
+ [Callable[[dict[str, typing.Any]], dict[str, typing.Any]]],
+ dict[str, typing.Any],
+ ]
+ mutate_config_and_state: Callable[
+ [
+ Callable[
+ [dict[str, typing.Any], dict[str, typing.Any]],
+ tuple[dict[str, typing.Any], dict[str, typing.Any]],
+ ]
+ ],
+ tuple[dict[str, typing.Any], dict[str, typing.Any]],
+ ]
persist_config_schema: Callable[[dict[str, typing.Any]], None]
@@ -74,6 +168,7 @@ class ExtensionPublication:
source_types_published: dict[str, type[SourceBase]]
resolvers_before: dict[ResolverType, type[Resolver]]
resolvers_published: dict[ResolverType, type[Resolver]]
+ public_http_claim: PublicHTTPRouteClaim | None = None
source_runtime_activation: SourceRuntimeActivation | None = None
restored: bool = False
@@ -104,6 +199,9 @@ def restore(self) -> None:
if self.source_runtime_activation is not None:
SourceManager.withdraw_runtime_activation(self.source_runtime_activation)
self.source_runtime_activation = None
+ if self.public_http_claim is not None:
+ self.public_http_claim.release()
+ self.public_http_claim = None
SourceManager.restore_source_types(
self.source_types_before,
self.source_types_published,
diff --git a/app/business/extension/state.py b/app/business/extension/state.py
index 11a5c09..b0371f9 100644
--- a/app/business/extension/state.py
+++ b/app/business/extension/state.py
@@ -17,7 +17,7 @@
from .errors import ExtensionNotInstalledError, ExtensionStateConflictError
-class ExtensionState(pydantic.BaseModel):
+class InstalledExtension(pydantic.BaseModel):
"""Stable Host-facing projection; SQLModel and table details remain private."""
model_config = pydantic.ConfigDict(frozen=True)
@@ -30,27 +30,52 @@ class ExtensionState(pydantic.BaseModel):
config_schema: dict[str, typing.Any] | None = None
-class ExtensionStateStore(typing.Protocol):
- def list(self) -> tuple[ExtensionState, ...]: ...
+StateMutation: typing.TypeAlias = Callable[[dict[str, typing.Any]], dict[str, typing.Any]]
+ConfigStateMutation: typing.TypeAlias = Callable[
+ [dict[str, typing.Any], dict[str, typing.Any]],
+ tuple[dict[str, typing.Any], dict[str, typing.Any]],
+]
- def get(self, name: str) -> ExtensionState | None: ...
- def install(self, name: str, version: str, nickname: str) -> ExtensionState: ...
+class ExtensionStore(typing.Protocol):
+ def list(self) -> tuple[InstalledExtension, ...]: ...
+
+ def get(self, name: str) -> InstalledExtension | None: ...
+
+ def install(self, name: str, version: str, nickname: str) -> InstalledExtension: ...
def uninstall(self, name: str) -> None: ...
- def update_config(self, name: str, config: dict[str, typing.Any]) -> ExtensionState: ...
+ def read_config(self, name: str) -> dict[str, typing.Any]: ...
+
+ def update_config(
+ self, name: str, config: dict[str, typing.Any]
+ ) -> InstalledExtension: ...
+
+ def read_state(self, name: str) -> dict[str, typing.Any]: ...
+
+ def mutate_state(
+ self,
+ name: str,
+ transform: StateMutation,
+ ) -> dict[str, typing.Any]: ...
+
+ def mutate_config_and_state(
+ self,
+ name: str,
+ transform: ConfigStateMutation,
+ ) -> tuple[dict[str, typing.Any], dict[str, typing.Any]]: ...
def update_config_schema(
self, name: str, schema: dict[str, typing.Any]
- ) -> ExtensionState: ...
+ ) -> InstalledExtension: ...
def set_peer_enabled(
self, name: str, peer_id: uuid.UUID, enabled: bool
- ) -> ExtensionState: ...
+ ) -> InstalledExtension: ...
-class SQLExtensionStateStore:
+class SQLExtensionStore:
"""Transactional adapter over the one canonical deployment relation."""
def __init__(
@@ -59,8 +84,8 @@ def __init__(
self._session_factory = session_factory
@staticmethod
- def _state(model: ExtensionModel) -> ExtensionState:
- return ExtensionState(
+ def _installed(model: ExtensionModel) -> InstalledExtension:
+ return InstalledExtension(
name=model.name,
version=model.version,
enabled=tuple(model.enabled),
@@ -71,17 +96,17 @@ def _state(model: ExtensionModel) -> ExtensionState:
),
)
- def list(self) -> tuple[ExtensionState, ...]:
+ def list(self) -> tuple[InstalledExtension, ...]:
with self._session_factory() as db:
rows = db.exec(sqlmodel.select(ExtensionModel).order_by(ExtensionModel.name)).all()
- return tuple(self._state(row) for row in rows)
+ return tuple(self._installed(row) for row in rows)
- def get(self, name: str) -> ExtensionState | None:
+ def get(self, name: str) -> InstalledExtension | None:
with self._session_factory() as db:
row = db.get(ExtensionModel, name)
- return self._state(row) if row is not None else None
+ return self._installed(row) if row is not None else None
- def install(self, name: str, version: str, nickname: str) -> ExtensionState:
+ def install(self, name: str, version: str, nickname: str) -> InstalledExtension:
with self._session_factory() as db:
locked = (
db.connection()
@@ -105,6 +130,7 @@ def install(self, name: str, version: str, nickname: str) -> ExtensionState:
enabled=[],
nickname=nickname,
config={},
+ state={},
config_schema=None,
)
elif row.version != version:
@@ -112,6 +138,10 @@ def install(self, name: str, version: str, nickname: str) -> ExtensionState:
raise ExtensionStateConflictError(
f"Cannot change {name} while one or more peers are enabled"
)
+ if row.state:
+ raise ExtensionStateConflictError(
+ f"Cannot change {name} while Extension state is not empty"
+ )
row.version = version
row.nickname = nickname
row.config_schema = None
@@ -120,7 +150,7 @@ def install(self, name: str, version: str, nickname: str) -> ExtensionState:
db.add(row)
db.commit()
db.refresh(row)
- return self._state(row)
+ return self._installed(row)
def uninstall(self, name: str) -> None:
with self._session_factory() as db:
@@ -141,7 +171,7 @@ def _update_json(
name: str,
field: typing.Literal["config", "config_schema"],
value: dict[str, typing.Any],
- ) -> ExtensionState:
+ ) -> InstalledExtension:
with self._session_factory() as db:
row = db.get(ExtensionModel, name)
if row is None:
@@ -150,19 +180,73 @@ def _update_json(
db.add(row)
db.commit()
db.refresh(row)
- return self._state(row)
+ return self._installed(row)
+
+ @staticmethod
+ def _locked_row(db: sqlmodel.Session, name: str) -> ExtensionModel:
+ row = db.exec(
+ sqlmodel.select(ExtensionModel).where(ExtensionModel.name == name).with_for_update()
+ ).one_or_none()
+ if row is None:
+ raise ExtensionNotInstalledError(f"{name} is not installed")
+ return row
+
+ def read_config(self, name: str) -> dict[str, typing.Any]:
+ with self._session_factory() as db:
+ row = db.get(ExtensionModel, name)
+ if row is None:
+ raise ExtensionNotInstalledError(f"{name} is not installed")
+ return dict(row.config)
- def update_config(self, name: str, config: dict[str, typing.Any]) -> ExtensionState:
+ def update_config(self, name: str, config: dict[str, typing.Any]) -> InstalledExtension:
return self._update_json(name, "config", config)
+ def read_state(self, name: str) -> dict[str, typing.Any]:
+ with self._session_factory() as db:
+ row = db.get(ExtensionModel, name)
+ if row is None:
+ raise ExtensionNotInstalledError(f"{name} is not installed")
+ return dict(row.state)
+
+ def mutate_state(
+ self,
+ name: str,
+ transform: StateMutation,
+ ) -> dict[str, typing.Any]:
+ with self._session_factory() as db:
+ row = self._locked_row(db, name)
+ updated = transform(dict(row.state))
+ if not isinstance(updated, dict):
+ raise TypeError("Extension state mutation must return an object")
+ row.state = dict(updated)
+ db.add(row)
+ db.commit()
+ return dict(row.state)
+
+ def mutate_config_and_state(
+ self,
+ name: str,
+ transform: ConfigStateMutation,
+ ) -> tuple[dict[str, typing.Any], dict[str, typing.Any]]:
+ with self._session_factory() as db:
+ row = self._locked_row(db, name)
+ config, state = transform(dict(row.config), dict(row.state))
+ if not isinstance(config, dict) or not isinstance(state, dict):
+ raise TypeError("Extension config/state mutation must return two objects")
+ row.config = dict(config)
+ row.state = dict(state)
+ db.add(row)
+ db.commit()
+ return dict(row.config), dict(row.state)
+
def update_config_schema(
self, name: str, schema: dict[str, typing.Any]
- ) -> ExtensionState:
+ ) -> InstalledExtension:
return self._update_json(name, "config_schema", schema)
def set_peer_enabled(
self, name: str, peer_id: uuid.UUID, enabled: bool
- ) -> ExtensionState:
+ ) -> InstalledExtension:
"""Use the shared atomic RPC; Core never performs array read-modify-write."""
statement = sqlalchemy.text(
f"SELECT * FROM {PROTOCOL_SCHEMA}.set_extension_peer_enabled("
@@ -185,4 +269,4 @@ def set_peer_enabled(
if row is None:
raise ExtensionNotInstalledError(f"{name} is not installed")
db.commit()
- return ExtensionState.model_validate(dict(row))
+ return InstalledExtension.model_validate(dict(row))
diff --git a/app/business/source/collect_job.py b/app/business/source/collect_job.py
index b406836..0a8b0dc 100644
--- a/app/business/source/collect_job.py
+++ b/app/business/source/collect_job.py
@@ -1,5 +1,7 @@
import datetime
+import json
+import sqlalchemy
import sqlmodel
from app.engine import SessionLocal
@@ -20,6 +22,39 @@
class SourceCollectJobManager:
"""Manager for source collect jobs."""
+ @classmethod
+ def ensure(
+ cls,
+ source_id: int,
+ config: dict,
+ ) -> tuple[SourceCollectJobModel, bool]:
+ """Atomically reuse or create one non-failed job for exact JSON input."""
+ canonical_config = json.dumps(config, sort_keys=True, separators=(",", ":"))
+ lock_identity = f"source-collect-job\0{source_id}\0{canonical_config}"
+ with SessionLocal() as db:
+ db.connection().execute(
+ sqlalchemy.text("SELECT pg_advisory_xact_lock(hashtextextended(:identity, 0))"),
+ {"identity": lock_identity},
+ )
+ jobs = db.exec(
+ sqlmodel.select(SourceCollectJobModel)
+ .where(
+ SourceCollectJobModel.source == source_id,
+ SourceCollectJobModel.status != SourceCollectJobStatus.FAILED,
+ )
+ .order_by(sqlmodel.col(SourceCollectJobModel.id))
+ .with_for_update()
+ ).all()
+ existing = next((job for job in jobs if dict(job.config) == config), None)
+ if existing is not None:
+ db.commit()
+ return existing, False
+ job = SourceCollectJobModel(source=source_id, config=dict(config))
+ db.add(job)
+ db.commit()
+ db.refresh(job)
+ return job, True
+
@classmethod
async def run(cls, job_id: SourceCollectJobID) -> bool:
"""Atomically claim and consume one pending job at most once."""
diff --git a/app/business/source/main.py b/app/business/source/main.py
index 09549d9..4efb454 100644
--- a/app/business/source/main.py
+++ b/app/business/source/main.py
@@ -11,6 +11,7 @@
from app.database_contract.profile import BUILTIN_SOURCE_TYPES_BY_ID
from app.schemas.info_base.block import BlockID
from app.schemas.source import (
+ CollectAt,
SourceCollectJobModel,
SourceModel,
SourceID,
@@ -82,6 +83,10 @@ async def record(self, data: typing.Any) -> None:
"""
raise NotImplementedError(f"{self.__class__.__name__} does not support passive record")
+ def scheduled_collect_config(self) -> dict[str, typing.Any]:
+ """Build immutable input for a scheduler-created collect job."""
+ return {}
+
def get_config(self) -> ConfigTV:
"""Get the configuration of the source."""
with SessionLocal() as db:
@@ -285,8 +290,12 @@ async def _run_scheduled_collect(cls, source_id: SourceID) -> None:
"""Create one durable collect job, then run the canonical job path."""
from .collect_job import SourceCollectJobManager
+ source = cls._get_source_ins(source_id)
with SessionLocal() as db:
- job = SourceCollectJobModel(source=source_id, config={})
+ job = SourceCollectJobModel(
+ source=source_id,
+ config=source.scheduled_collect_config(),
+ )
db.add(job)
db.commit()
db.refresh(job)
@@ -339,3 +348,47 @@ def create(cls, type_: str, nickname: Opt[str] = None) -> SourceModel:
cls._SOURCE_ROW_TYPES[source.id] = type_
return source
+
+ @classmethod
+ def ensure_exists(
+ cls,
+ type_: str,
+ *,
+ nickname: Opt[str] = None,
+ config: dict | None = None,
+ collect_at: CollectAt | None = None,
+ ) -> tuple[SourceModel, bool]:
+ """Return any existing Source of this type, or atomically create one.
+
+ This is an at-least-one primitive, not a uniqueness policy: existing Sources
+ are never renamed or rescheduled, and callers may still create more Sources.
+ """
+ with SessionLocal() as db:
+ db.connection().execute(
+ sqlalchemy.text("SELECT pg_advisory_xact_lock(hashtextextended(:type, 0))"),
+ {"type": type_},
+ )
+ source = db.exec(
+ sqlmodel.select(SourceModel)
+ .where(SourceModel.type == type_)
+ .order_by(sqlmodel.col(SourceModel.id))
+ .limit(1)
+ .with_for_update()
+ ).one_or_none()
+ created = source is None
+ if source is None:
+ source = SourceModel(
+ type=type_,
+ nickname=nickname,
+ config=dict(config or {}),
+ collect_at=collect_at,
+ )
+ db.add(source)
+ db.commit()
+ db.refresh(source)
+
+ if source.id is not None:
+ cls._SOURCE_ROW_TYPES[source.id] = type_
+ if created and source.collect_at is not None:
+ cls.set_up_collect_jobs({type_})
+ return source, created
diff --git a/app/database_contract/constants.py b/app/database_contract/constants.py
index f7db657..9ec144f 100644
--- a/app/database_contract/constants.py
+++ b/app/database_contract/constants.py
@@ -4,7 +4,7 @@
CONTRACT_FORMAT = 1
-CONTRACT_REVISION = "peer-database-runtime-v2"
+CONTRACT_REVISION = "peer-database-runtime-v3"
PROTOCOL_SCHEMA = "inkcre"
INTERNAL_SCHEMA = "inkcre_internal"
diff --git a/app/middleware.py b/app/middleware.py
index 7ffd271..8e585ef 100644
--- a/app/middleware.py
+++ b/app/middleware.py
@@ -19,6 +19,7 @@
)
from app.settings import settings
from libs.obsrv.main import get_logger
+from app.business.extension.runtime import PublicHTTPRouteClaim
def decode_peer_jwt(
@@ -95,7 +96,6 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
extra={
"method": request.method,
"path": request.url.path,
- "query_params": str(request.query_params),
},
)
@@ -161,6 +161,7 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
request.url.path in {"/heartbeat", "/livez", "/readyz"}
or request.url.path == "/docs"
or request.url.path.startswith("/openapi.json")
+ or PublicHTTPRouteClaim.permits(request.method, request.url.path)
):
return await call_next(request)
diff --git a/app/routes/extension.py b/app/routes/extension.py
index f0c9c08..e1c835a 100644
--- a/app/routes/extension.py
+++ b/app/routes/extension.py
@@ -5,7 +5,7 @@
import fastapi
import pydantic
-from app.business.extension import EXTENSION_HOST, ExtensionState
+from app.business.extension import EXTENSION_HOST, InstalledExtension
from app.business.extension.errors import (
ExtensionAcquisitionError,
ExtensionCompatibilityError,
@@ -47,12 +47,12 @@ def _raise_http_error(error: ExtensionHostError) -> typing.NoReturn:
@ROUTER.get("")
-def list_extensions() -> tuple[ExtensionState, ...]:
+def list_extensions() -> tuple[InstalledExtension, ...]:
return EXTENSION_HOST.list()
@ROUTER.get("/{namespace}/{name}")
-def get_extension(namespace: str, name: str) -> ExtensionState:
+def get_extension(namespace: str, name: str) -> InstalledExtension:
try:
return EXTENSION_HOST.get(_coordinate(namespace, name))
except ExtensionHostError as error:
@@ -64,7 +64,7 @@ def install_extension(
namespace: str,
name: str,
version: str = fastapi.Query(...),
-) -> ExtensionState:
+) -> InstalledExtension:
"""Install one exact published Extension Release with no enabled peers."""
try:
return EXTENSION_HOST.install(_coordinate(namespace, name), version)
@@ -86,7 +86,7 @@ def update_extension_config(
namespace: str,
name: str,
body: dict[str, typing.Any] = fastapi.Body(...),
-) -> ExtensionState:
+) -> InstalledExtension:
try:
return EXTENSION_HOST.update_config(_coordinate(namespace, name), body)
except pydantic.ValidationError as error:
@@ -99,7 +99,7 @@ def update_extension_config(
@ROUTER.post("/{namespace}/{name}/enable")
-async def enable_extension(namespace: str, name: str) -> ExtensionState:
+async def enable_extension(namespace: str, name: str) -> InstalledExtension:
try:
return await EXTENSION_HOST.enable(_coordinate(namespace, name))
except ExtensionHostError as error:
@@ -107,7 +107,7 @@ async def enable_extension(namespace: str, name: str) -> ExtensionState:
@ROUTER.post("/{namespace}/{name}/disable")
-async def disable_extension(namespace: str, name: str) -> ExtensionState:
+async def disable_extension(namespace: str, name: str) -> InstalledExtension:
try:
return await EXTENSION_HOST.disable(_coordinate(namespace, name))
except ExtensionHostError as error:
diff --git a/app/schemas/extension/main.py b/app/schemas/extension/main.py
index cd4c0cc..ce5850e 100644
--- a/app/schemas/extension/main.py
+++ b/app/schemas/extension/main.py
@@ -31,6 +31,10 @@ class ExtensionModel(sqlmodel.SQLModel, table=True):
f"version ~ '{EXTENSION_SEMVER_PATTERN}'",
name="extensions_version_canonical",
),
+ sqlalchemy.CheckConstraint(
+ "jsonb_typeof(state) = 'object'",
+ name="extensions_state_object",
+ ),
)
name: ExtensionName = sqlmodel.Field(
@@ -61,6 +65,14 @@ class ExtensionModel(sqlmodel.SQLModel, table=True):
nullable=False,
),
)
+ state: dict = sqlmodel.Field(
+ default_factory=dict,
+ sa_column=sqlalchemy.Column(
+ sqlalchemy.dialects.postgresql.JSONB,
+ server_default=sqlalchemy.text("'{}'::jsonb"),
+ nullable=False,
+ ),
+ )
config_schema: dict | None = sqlmodel.Field(
default=None,
sa_column=sqlalchemy.Column(
diff --git a/app/version.py b/app/version.py
index 52515b3..6ef8d87 100644
--- a/app/version.py
+++ b/app/version.py
@@ -1,3 +1,3 @@
"""Single source of truth for the Core and Core Extension Host SDK version."""
-CORE_VERSION = "0.1.0"
+CORE_VERSION = "0.1.1"
diff --git a/deploy/profiles/production.json b/deploy/profiles/production.json
index dd5fe25..509c3ec 100644
--- a/deploy/profiles/production.json
+++ b/deploy/profiles/production.json
@@ -2,8 +2,8 @@
"format": 1,
"environment": "production",
"database_contract": {
- "revision": "peer-database-runtime-v2",
- "migration_head": "b8c1d2e3f4a5",
+ "revision": "peer-database-runtime-v3",
+ "migration_head": "c7d8e9f0a1b2",
"protocol_schema": "inkcre"
},
"client": {
diff --git a/extensions/twitter/__init__.py b/extensions/twitter/__init__.py
index 6bb098f..4fac452 100644
--- a/extensions/twitter/__init__.py
+++ b/extensions/twitter/__init__.py
@@ -2,10 +2,12 @@
import sqlmodel
from typing import Optional as Opt
from fastapi import APIRouter
-from app.business.extension.main import ExtensionBase
+from app.business.extension.main import ExtensionBase, PublicHTTPRoute
from app.business.info_base.resolver import ResolverManager
from app.business.source import SourceManager
+from .setup_flow import TwitterExtensionState
+
class TwitterExtensionConfig(sqlmodel.SQLModel):
# TODO move to SourceConfig
@@ -24,7 +26,37 @@ class Extension(
ExtensionBase[TwitterExtensionConfig],
ext_id="twitter",
config_cls=TwitterExtensionConfig,
+ state_cls=TwitterExtensionState,
):
+ @classmethod
+ def update_config(
+ cls,
+ new_config: dict[str, typing.Any] | TwitterExtensionConfig,
+ ) -> TwitterExtensionConfig:
+ from .setup_flow import TwitterExtensionState, _fingerprint, _terminal
+
+ updated = (
+ TwitterExtensionConfig.model_validate(new_config)
+ if isinstance(new_config, dict)
+ else new_config
+ )
+
+ def replace(config_model, state_model):
+ current_config = TwitterExtensionConfig.model_validate(config_model.model_dump())
+ state = TwitterExtensionState.model_validate(state_model.model_dump())
+ if _fingerprint(current_config) != _fingerprint(updated):
+ state.account = None
+ state.oauth_transactions = {
+ key: _terminal(value, "expired", error="OAuth App changed")
+ if value.status in {"pending", "exchanging"}
+ else value
+ for key, value in state.oauth_transactions.items()
+ }
+ return updated, state
+
+ config, _ = cls.mutate_config_and_state(replace)
+ return config
+
@classmethod
def _init_resolvers(cls):
from .resolver import TweetResolver
@@ -57,11 +89,10 @@ async def on_close(cls):
@classmethod
def _register_apis(cls, router: APIRouter):
- from .api import TwitterAPI
+ from .setup_flow import register_setup_routes
- TwitterAPI.new(api_router=router)
- router.post("/bookmark")(
- lambda nickname: SourceManager.create(
- f"extensions.{cls.__extid__}.bookmark.Source", nickname
- )
- )
+ register_setup_routes(router)
+
+ @classmethod
+ def public_http_routes(cls) -> tuple[PublicHTTPRoute, ...]:
+ return (PublicHTTPRoute(method="GET", path="/auth/callback"),)
diff --git a/extensions/twitter/api.py b/extensions/twitter/api.py
index f24a01b..c68a2a9 100644
--- a/extensions/twitter/api.py
+++ b/extensions/twitter/api.py
@@ -1,21 +1,17 @@
import abc
import asyncio
-import base64
import datetime
-import os
from pathlib import Path
import re
import typing
-import secrets
-import aiohttp
-import fastapi
+from authlib.integrations.httpx_client import AsyncOAuth2Client # pyrefly: ignore[untyped-import]
+from authlib.integrations.base_client.errors import OAuthError # pyrefly: ignore[untyped-import]
+import httpx
import sqlmodel
import twikit
import twikit.media
-import urllib.parse
from typing import Optional as Opt
from dd import dd
-from utils.base import AIOHTTP_CONNECTOR_GETTER
from utils.datetime_ import get_timestamp
from .schema import TweetPhoto, TweetVideo, VideoVariant
from . import Extension
@@ -63,39 +59,32 @@ async def close_singleton(cls) -> None:
cls.SINGLETON = None
@classmethod
- def new(cls, api_router: Opt[fastapi.APIRouter] = None) -> "TwitterAPI":
+ def new(
+ cls,
+ *,
+ expected_authorization_id: str | None = None,
+ ) -> "TwitterAPI":
"""Create an instance of the Twitter API client.
Use `config.backend` to determine which backend to use.
"""
+ config = Extension.get_config()
+ backend_type = config.backend
+ if backend_type == "official":
+ return OfficialAPI.from_extension(expected_authorization_id=expected_authorization_id)
if cls.SINGLETON is not None:
return cls.SINGLETON
- else:
- backend_type = Extension.config.backend
- if backend_type == "official":
- cls.SINGLETON = OfficialAPI(
- client_id=Extension.config.client_id,
- client_secret=Extension.config.client_secret,
- )
- if api_router:
- api_router.get("/auth/authorize")(cls.SINGLETON.get_oauth_authorize_url)
- api_router.get("/auth/callback")(cls.SINGLETON.handle_oauth_callback)
- else:
- # log warning
- pass
- elif backend_type == "twikit":
- cls.SINGLETON = TwikitAPI(
- email=Extension.config.email,
- username=Extension.config.username,
- password=Extension.config.password,
- totp_secret=Extension.config.totp_secret,
- language=Extension.config.api_language,
- proxy=Extension.config.proxy,
- )
- else:
- raise ValueError(f"Unknown backend type: {backend_type}")
-
+ if backend_type == "twikit":
+ cls.SINGLETON = TwikitAPI(
+ email=config.email,
+ username=config.username,
+ password=config.password,
+ totp_secret=config.totp_secret,
+ language=config.api_language,
+ proxy=config.proxy,
+ )
return cls.SINGLETON
+ raise ValueError(f"Unknown backend type: {backend_type}")
async def close(self): ...
@@ -126,8 +115,6 @@ async def get_replies(
class OfficialAPI(TwitterAPI):
"""Official Twitter API client."""
- state = None
- challenge = None
request_records: dict[str, tuple[int, datetime.datetime]] = {}
"""How many requests made to each endpoint for last 15 mins.
"""
@@ -135,13 +122,53 @@ class OfficialAPI(TwitterAPI):
"""When the rate limit for each endpoint will reset.
"""
- def __init__(self, client_id: str, client_secret: str):
+ def __init__( # noqa: PLR0913
+ self,
+ client_id: str,
+ client_secret: str,
+ *,
+ token: dict[str, typing.Any],
+ user_id: str,
+ user_handle: str,
+ authorization_id: str,
+ ):
self.__client_id = client_id
self.__client_secret = client_secret
- self.__access_token: Opt[str] = None
- self.__refresh_token: Opt[str] = None
- self.__user_id: Opt[str] = None
- self.__user_handle: Opt[str] = None
+ self.__token = token
+ self.__user_id = user_id
+ self.__user_handle = user_handle
+ self.__authorization_id = authorization_id
+
+ @classmethod
+ def from_extension(
+ cls,
+ *,
+ expected_authorization_id: str | None = None,
+ ) -> "OfficialAPI":
+ from .setup_flow import TwitterExtensionState, TwitterSetupConflict, _fingerprint
+
+ config = Extension.get_config()
+ state = TwitterExtensionState.model_validate(Extension.get_state().model_dump())
+ account = state.account
+ if (
+ account is None
+ or account.reconnect_required
+ or account.app_fingerprint != _fingerprint(config)
+ ):
+ raise TwitterSetupConflict("Twitter account is not connected")
+ if (
+ expected_authorization_id is not None
+ and account.authorization_id != expected_authorization_id
+ ):
+ raise TwitterSetupConflict("Twitter authorization changed before collection")
+ return cls(
+ config.client_id,
+ config.client_secret,
+ token=dict(account.token),
+ user_id=account.user_id,
+ user_handle=account.handle,
+ authorization_id=account.authorization_id,
+ )
@property
def user_handle(self) -> str:
@@ -182,14 +209,20 @@ async def _request( # noqa: PLR0913
- Error handling
- Resopnse body parsing
"""
- if not self.__access_token:
- # TODO raise Unauthorized
- raise ValueError("Access token is not set. Please authorize first.")
-
+ from .setup_flow import TwitterExtensionState, TwitterSetupConflict, _fingerprint
+
+ latest = TwitterExtensionState.model_validate(Extension.get_state().model_dump())
+ latest_config = Extension.get_config()
+ if (
+ latest.account is None
+ or latest.account.reconnect_required
+ or latest.account.authorization_id != self.__authorization_id
+ or latest.account.app_fingerprint != _fingerprint(latest_config)
+ ):
+ raise TwitterSetupConflict("Twitter authorization changed before provider access")
+ self.__token = dict(latest.account.token)
+ request_token = dict(self.__token)
endpoint_with_params = endpoint.format(**path_params) if path_params else endpoint
- headers = {
- "Authorization": f"Bearer {self.__access_token}",
- }
rate_limit_reset_at = self.rate_limit_reset.get(endpoint)
if rate_limit_reset_at:
@@ -214,17 +247,60 @@ async def _request( # noqa: PLR0913
# else:
# last_request_count, last_15m_start_at = 0, datetime.datetime.now()
- async with aiohttp.ClientSession(connector=AIOHTTP_CONNECTOR_GETTER()) as session:
- async with session.request(
+ async def update_token(
+ token: dict[str, typing.Any],
+ **_: typing.Any,
+ ) -> None:
+ from .setup_flow import TwitterExtensionState, TwitterSetupConflict
+
+ def update(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ if (
+ state.account is None or state.account.authorization_id != self.__authorization_id
+ ):
+ raise TwitterSetupConflict("Twitter authorization changed during refresh")
+ if dict(state.account.token) != request_token:
+ raise TwitterSetupConflict("Twitter token changed during refresh")
+ state.account.token = dict(token)
+ return state
+
+ Extension.mutate_state(update)
+ self.__token = dict(token)
+
+ def require_reconnect() -> None:
+ from .setup_flow import TwitterExtensionState
+
+ def update(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ if (
+ state.account is not None
+ and state.account.authorization_id == self.__authorization_id
+ ):
+ state.account.reconnect_required = True
+ return state
+
+ Extension.mutate_state(update)
+
+ client = AsyncOAuth2Client(
+ client_id=self.__client_id,
+ client_secret=self.__client_secret,
+ token=self.__token,
+ token_endpoint="https://api.x.com/2/oauth2/token",
+ token_endpoint_auth_method="client_secret_basic",
+ update_token=update_token,
+ timeout=10,
+ )
+ try:
+ response = await client.request(
method,
f"https://api.x.com/2{endpoint_with_params}",
params=query,
- headers=headers,
- ) as resp:
- if resp.status == 429:
- x_rate_limit_reset = resp.headers.get("x-rate-limit-reset")
- if x_rate_limit_reset:
- self.rate_limit_reset[endpoint] = int(x_rate_limit_reset)
+ json=body,
+ )
+ if response.status_code == 429:
+ x_rate_limit_reset = response.headers.get("x-rate-limit-reset")
+ if x_rate_limit_reset:
+ self.rate_limit_reset[endpoint] = int(x_rate_limit_reset)
# # Rate limit exceeded but not expected, set request count to max
# # and request again when rate limit reset
@@ -233,15 +309,32 @@ async def _request( # noqa: PLR0913
# datetime.datetime.now()
# )
- if retried < 3:
- return await self._request(
- method, endpoint, path_params, query, body, retried + 1
- )
- else:
- raise RuntimeError("Twitter API rate limit exceeded after retries")
-
- resp.raise_for_status()
- return await resp.json()
+ if retried < 3:
+ return await self._request(
+ method, endpoint, path_params, query, body, retried + 1
+ )
+ raise RuntimeError("Twitter API rate limit exceeded after retries")
+
+ response.raise_for_status()
+ payload = response.json()
+ if not isinstance(payload, dict):
+ raise RuntimeError("Twitter API returned an invalid response")
+ return payload
+ except OAuthError:
+ require_reconnect()
+ raise RuntimeError("Twitter authorization requires reconnection") from None
+ except httpx.HTTPStatusError as error:
+ if error.response.status_code == 401:
+ require_reconnect()
+ raise RuntimeError("Twitter authorization requires reconnection") from None
+ raise RuntimeError("Twitter API request failed") from error
+ except httpx.HTTPError as error:
+ raise RuntimeError("Twitter API request failed") from error
+ finally:
+ await typing.cast(httpx.AsyncClient, client).aclose()
+
+ async def close(self) -> None:
+ """Official clients are operation-scoped and hold no open transport."""
async def get_user(self) -> tuple[str, str]:
"""Get the user info the token represents and store to state.
@@ -398,112 +491,6 @@ async def get_replies(
)
return res
- @staticmethod
- def _get_oauth_redirect_url():
- return os.getenv("API_BASE_URL", "") + "/twitter/auth/callback"
-
- def get_oauth_authorize_url(self) -> str:
- BASE_URL = "https://x.com/i/oauth2/authorize"
- REDIRECT_URL = self._get_oauth_redirect_url()
- CLIENT_ID = self.__client_id
-
- scope = " ".join(
- [
- "tweet.read",
- "users.read",
- "bookmark.read",
- "bookmark.write",
- "offline.access",
- ]
- )
- self.state = secrets.token_urlsafe(16) # A random string to prevent CSRF attacks
- self.code_challenge = secrets.token_urlsafe(32) # Code challenge for PKCE
-
- params = {
- "response_type": "code",
- "client_id": CLIENT_ID,
- "state": self.state,
- "code_challenge": self.code_challenge,
- "code_challenge_method": "plain",
- "redirect_uri": REDIRECT_URL,
- "scope": scope,
- }
-
- return BASE_URL + "?" + urllib.parse.urlencode(params)
-
- async def handle_oauth_callback(self, code: str, state: str):
- """Handle the OAuth2 callback from Twitter.
-
- Exchange the authorization code for an access token.
- """
- # verify state
- if state != self.state:
- raise ValueError("Invalid state parameter")
-
- token_url = "https://api.x.com/2/oauth2/token" # noqa: S105
- CLIENT_ID = self.__client_id
- CLIENT_SECRET = self.__client_secret
-
- data = {
- "grant_type": "authorization_code",
- "code": code,
- "redirect_uri": self._get_oauth_redirect_url(),
- "client_id": CLIENT_ID,
- "code_verifier": self.code_challenge,
- }
-
- headers = {
- "Content-Type": "application/x-www-form-urlencoded",
- "Authorization": f"Basic {
- base64.b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}'.encode()).decode()
- }",
- }
-
- async with aiohttp.ClientSession(connector=AIOHTTP_CONNECTOR_GETTER()) as session:
- async with session.post(token_url, data=data, headers=headers) as resp:
- resp.raise_for_status()
- resp_body = await resp.json()
- access_token = resp_body.get("access_token")
- refresh_token = resp_body.get("refresh_token")
- if not access_token or not refresh_token:
- raise ValueError("Failed to obtain access token or refresh token")
- # Extension.state["access_token"] = access_token
- # Extension.state["refresh_token"] = refresh_token
- self.__access_token = access_token
- self.__refresh_token = refresh_token
-
- self.state = None
- self.challenge = None
-
- # Get user info and store to state
- await self.get_user()
-
- return resp_body
-
- async def refresh_access_token(self, refresh_token: str) -> str:
- """Get a new access token using the refresh token.
-
- Docs https://docs.x.com/fundamentals/authentication/oauth-2-0/authorization-code#refresh-tokens
- """
- token_url = "https://api.x.com/2/oauth2/token" # noqa: S105
- CLIENT_ID = self.__client_id
-
- data = {
- "grant_type": "refresh_token",
- "refresh_token": refresh_token,
- "client_id": CLIENT_ID,
- }
-
- headers = {
- "Content-Type": "application/x-www-form-urlencoded",
- }
-
- async with aiohttp.ClientSession(connector=AIOHTTP_CONNECTOR_GETTER()) as session:
- async with session.post(token_url, data=data, headers=headers) as resp:
- resp.raise_for_status()
- token_response = await resp.json()
- return token_response
-
class TwikitAPI(TwitterAPI):
"""Twikit API client."""
diff --git a/extensions/twitter/bookmark.py b/extensions/twitter/bookmark.py
index d03ee18..debd905 100644
--- a/extensions/twitter/bookmark.py
+++ b/extensions/twitter/bookmark.py
@@ -29,6 +29,19 @@ class Source(SourceBase[SourceConfig], config_cls=SourceConfig):
API_BASE_URL = "https://api.x.com/2"
+ def scheduled_collect_config(self) -> dict[str, typing.Any]:
+ from .setup_flow import TwitterExtensionState, TwitterSetupConflict
+ from . import Extension
+
+ state = TwitterExtensionState.model_validate(Extension.get_state().model_dump())
+ if state.account is None or state.account.reconnect_required:
+ raise TwitterSetupConflict("Twitter account is not connected")
+ return {
+ "full": False,
+ "result_limit": 40,
+ "authorization_id": state.account.authorization_id,
+ }
+
async def collect(self, job: "SourceCollectJobModel") -> None:
"""Collect all new bookmarks and its notes.
@@ -45,11 +58,17 @@ async def collect(self, job: "SourceCollectJobModel") -> None:
config = job.config or {}
full = config.get("full", False)
result_limit = config.get("result_limit", 40)
+ authorization_id = config.get("authorization_id")
+ if not isinstance(authorization_id, str) or not authorization_id:
+ raise RuntimeError("Twitter collect job has no authorization identity")
page = job.state.get("page") if job.state else None
- api_client = TwitterAPI.new()
- bookmarks_res = await api_client.get_bookmarks(page=page, max_results=result_limit)
+ api_client = TwitterAPI.new(expected_authorization_id=authorization_id)
+ try:
+ bookmarks_res = await api_client.get_bookmarks(page=page, max_results=result_limit)
+ finally:
+ await api_client.close()
# find new tweets start point
old_start_at = len(bookmarks_res.tweets)
@@ -100,7 +119,7 @@ async def collect(self, job: "SourceCollectJobModel") -> None:
)
)
- if not full:
+ if not full and bookmarks_res.tweets:
state = self.get_state()
state["latest_tweet_id"] = bookmarks_res.tweets[0].id
self.set_state(state)
diff --git a/extensions/twitter/pyproject.toml b/extensions/twitter/pyproject.toml
index 80e6418..3e72116 100644
--- a/extensions/twitter/pyproject.toml
+++ b/extensions/twitter/pyproject.toml
@@ -4,13 +4,15 @@ build-backend = "setuptools.build_meta"
[project]
name = "inkcre-ext-twitter"
-version = "0.1.1"
+version = "0.2.0"
description = "Twitter extension for InKCre"
authors = [{ name = "Lan_zhijiang", email = "lanzhijiang@foxmail.com" }]
dependencies = [
+ "authlib>=1.7.2,<2.0.0",
"aiohttp>=3.14.1,<4.0.0",
"datadot>=1.0.0,<2.0.0",
"fastapi>=0.139.2,<0.140.0",
+ "httpx>=0.28.1,<0.29.0",
"sqlmodel>=0.0.24,<0.0.25",
"twikit>=2.3.3,<3.0.0",
]
@@ -24,7 +26,7 @@ twitter = "extensions.twitter:Extension"
name = "inkcre/twitter"
nickname = "Twitter"
host-sdk = "core-py"
-host-sdk-version = ">=0.1.0 <0.2.0"
+host-sdk-version = ">=0.1.1 <0.2.0"
[tool.setuptools]
packages = ["extensions.twitter"]
diff --git a/extensions/twitter/setup_flow.py b/extensions/twitter/setup_flow.py
new file mode 100644
index 0000000..acb8f4d
--- /dev/null
+++ b/extensions/twitter/setup_flow.py
@@ -0,0 +1,703 @@
+"""Whole-Extension setup workflow for the Twitter Extension."""
+
+from __future__ import annotations
+
+import base64
+import datetime
+import hashlib
+import html
+import secrets
+import typing
+import uuid
+
+from authlib.integrations.httpx_client import ( # pyrefly: ignore[untyped-import]
+ AsyncOAuth2Client,
+ OAuth2Client,
+)
+from authlib.integrations.base_client.errors import OAuthError # pyrefly: ignore[untyped-import]
+import fastapi
+from fastapi.responses import HTMLResponse
+import httpx
+import sqlmodel
+
+from app.business.source import SourceManager
+from app.engine import SessionLocal
+from app.schemas.source import (
+ CollectAt,
+ SourceCollectJobModel,
+ SourceCollectJobStatus,
+ SourceModel,
+)
+
+
+if typing.TYPE_CHECKING:
+ from . import TwitterExtensionConfig
+
+
+AUTHORIZE_URL = "https://x.com/i/oauth2/authorize"
+TOKEN_URL = "https://api.x.com/2/oauth2/token" # noqa: S105
+CURRENT_USER_URL = "https://api.x.com/2/users/me"
+SCOPES = ("tweet.read", "users.read", "bookmark.read", "offline.access")
+TRANSACTION_LIFETIME = datetime.timedelta(minutes=10)
+TERMINAL_RETENTION = datetime.timedelta(minutes=10)
+MAX_TRANSACTIONS = 8
+BOOKMARK_SOURCE_TYPE = "extensions.twitter.bookmark.Source"
+
+
+def _now() -> datetime.datetime:
+ return datetime.datetime.now(datetime.UTC)
+
+
+class TwitterAccount(sqlmodel.SQLModel):
+ token: dict[str, typing.Any]
+ user_id: str
+ handle: str
+ scopes: tuple[str, ...] = ()
+ app_fingerprint: str
+ authorization_id: str
+ connected_at: datetime.datetime
+ reconnect_required: bool = False
+
+
+class OAuthTransaction(sqlmodel.SQLModel):
+ status: typing.Literal["pending", "exchanging", "succeeded", "failed", "expired"]
+ provider_state: str | None = None
+ pkce_verifier: str | None = None
+ app_fingerprint: str
+ redirect_uri: str
+ created_at: datetime.datetime
+ expires_at: datetime.datetime
+ closed_at: datetime.datetime | None = None
+ error: str | None = None
+
+
+class TwitterExtensionState(sqlmodel.SQLModel):
+ account: TwitterAccount | None = None
+ oauth_transactions: dict[str, OAuthTransaction] = sqlmodel.Field(default_factory=dict)
+ bookmark_source_id: int | None = None
+
+
+class OAuthAppInput(sqlmodel.SQLModel):
+ client_id: str = sqlmodel.Field(min_length=1, max_length=256)
+ client_secret: str = sqlmodel.Field(min_length=1, max_length=1024)
+ confirm_account_reset: bool = False
+
+
+class BookmarkSourceInput(sqlmodel.SQLModel):
+ source_id: int | None = None
+ nickname: str = sqlmodel.Field(default="Twitter Bookmarks", min_length=1, max_length=120)
+ collect_at: CollectAt = sqlmodel.Field(default_factory=CollectAt)
+
+
+class OAuthTransactionView(sqlmodel.SQLModel):
+ id: str
+ status: str
+ authorize_url: str | None = None
+ expires_at: datetime.datetime
+ error: str | None = None
+
+
+class TwitterSetupStatus(sqlmodel.SQLModel):
+ backend: str
+ callback_url: str
+ oauth_app_configured: bool
+ client_id: str | None = None
+ connected: bool
+ user_id: str | None = None
+ handle: str | None = None
+ scopes: tuple[str, ...] = ()
+ reconnect_required: bool = False
+ bookmark_source_id: int | None = None
+ bookmark_sources: tuple[BookmarkSourceInput, ...] = ()
+ bookmark_source_ready: bool = False
+ ready: bool = False
+
+
+class TwitterSetupError(RuntimeError): ...
+
+
+class TwitterSetupConflict(TwitterSetupError): ...
+
+
+class TwitterProviderError(TwitterSetupError): ...
+
+
+def _extension():
+ from . import Extension
+
+ return Extension
+
+
+def _state() -> TwitterExtensionState:
+ return TwitterExtensionState.model_validate(_extension().get_state().model_dump())
+
+
+def _config() -> TwitterExtensionConfig:
+ return _extension().get_config()
+
+
+def _fingerprint(config: TwitterExtensionConfig) -> str:
+ material = (
+ f"inkcre-twitter-oauth-app\0{config.client_id}\0{config.client_secret}".encode()
+ )
+ return hashlib.sha256(material).hexdigest()
+
+
+def _redirect_uri() -> str:
+ from app.settings import settings
+
+ base = (settings.client_base_url or "").rstrip("/")
+ if not base:
+ raise TwitterSetupError("Core API base URL is not configured")
+ return f"{base}/twitter/auth/callback"
+
+
+def _pkce_pair() -> tuple[str, str]:
+ verifier = secrets.token_urlsafe(48)
+ digest = hashlib.sha256(verifier.encode()).digest()
+ challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
+ return verifier, challenge
+
+
+def _bounded_provider_error(error: BaseException | str) -> str:
+ value = str(error).replace("\r", " ").replace("\n", " ").strip()
+ return (value or "Twitter authorization failed")[:240]
+
+
+def _transaction_view(
+ transaction_id: str,
+ transaction: OAuthTransaction,
+ *,
+ authorize_url: str | None = None,
+) -> OAuthTransactionView:
+ return OAuthTransactionView(
+ id=transaction_id,
+ status=transaction.status,
+ authorize_url=authorize_url,
+ expires_at=transaction.expires_at,
+ error=transaction.error,
+ )
+
+
+def _terminal(
+ transaction: OAuthTransaction,
+ status: typing.Literal["succeeded", "failed", "expired"],
+ *,
+ error: str | None = None,
+) -> OAuthTransaction:
+ return transaction.model_copy(
+ update={
+ "status": status,
+ "provider_state": None,
+ "pkce_verifier": None,
+ "closed_at": _now(),
+ "error": error,
+ }
+ )
+
+
+def _source_and_job_status(
+ state: TwitterExtensionState,
+) -> tuple[int | None, tuple[BookmarkSourceInput, ...], bool]:
+ source_id = state.bookmark_source_id
+ account = state.account
+ with SessionLocal() as db:
+ sources = db.exec(
+ sqlmodel.select(SourceModel)
+ .where(SourceModel.type == BOOKMARK_SOURCE_TYPE)
+ .order_by(sqlmodel.col(SourceModel.id))
+ ).all()
+ source_views = tuple(
+ BookmarkSourceInput(
+ source_id=source.id,
+ nickname=source.nickname or "Twitter Bookmarks",
+ collect_at=source.collect_at or CollectAt(),
+ )
+ for source in sources
+ )
+ if source_id is None or account is None:
+ return source_id, source_views, False
+ source = next((item for item in sources if item.id == source_id), None)
+ if source is None:
+ return None, source_views, False
+ jobs = db.exec(
+ sqlmodel.select(SourceCollectJobModel)
+ .where(SourceCollectJobModel.source == source_id)
+ .order_by(sqlmodel.col(SourceCollectJobModel.id).desc())
+ ).all()
+ ready = any(
+ job.status != SourceCollectJobStatus.FAILED
+ and (job.config or {}).get("authorization_id") == account.authorization_id
+ for job in jobs
+ )
+ return source_id, source_views, ready
+
+
+def get_setup_status() -> TwitterSetupStatus:
+ config = _config()
+ state = _state()
+ source_id, sources, source_ready = _source_and_job_status(state)
+ configured = bool(config.client_id and config.client_secret)
+ account = state.account
+ connected = (
+ account is not None
+ and configured
+ and not account.reconnect_required
+ and account.app_fingerprint == _fingerprint(config)
+ )
+ return TwitterSetupStatus(
+ backend=config.backend,
+ callback_url=_redirect_uri(),
+ oauth_app_configured=configured,
+ client_id=config.client_id or None,
+ connected=connected,
+ user_id=account.user_id if connected and account is not None else None,
+ handle=account.handle if connected and account is not None else None,
+ scopes=account.scopes if connected and account is not None else (),
+ reconnect_required=bool(account and account.reconnect_required),
+ bookmark_source_id=source_id,
+ bookmark_sources=sources,
+ bookmark_source_ready=source_ready,
+ ready=connected and source_ready,
+ )
+
+
+def save_oauth_app(body: OAuthAppInput) -> TwitterSetupStatus:
+ extension = _extension()
+
+ def update(config_model, state_model):
+ from . import TwitterExtensionConfig
+
+ config = TwitterExtensionConfig.model_validate(config_model.model_dump())
+ state = TwitterExtensionState.model_validate(state_model.model_dump())
+ old_fingerprint = _fingerprint(config)
+ next_config = config.model_copy(
+ update={
+ "backend": "official",
+ "client_id": body.client_id.strip(),
+ "client_secret": body.client_secret,
+ }
+ )
+ fingerprint_changed = _fingerprint(next_config) != old_fingerprint
+ has_live_setup = state.account is not None or any(
+ value.status in {"pending", "exchanging"}
+ for value in state.oauth_transactions.values()
+ )
+ if fingerprint_changed and has_live_setup and not body.confirm_account_reset:
+ raise TwitterSetupConflict(
+ "Replacing the OAuth App requires confirmation because it disconnects the account"
+ )
+ config.backend = "official"
+ config.client_id = body.client_id.strip()
+ config.client_secret = body.client_secret
+ if fingerprint_changed:
+ state.account = None
+ state.oauth_transactions = {
+ key: _terminal(value, "expired", error="OAuth App changed")
+ if value.status in {"pending", "exchanging"}
+ else value
+ for key, value in state.oauth_transactions.items()
+ }
+ return config, state
+
+ extension.mutate_config_and_state(update)
+ return get_setup_status()
+
+
+def begin_oauth() -> OAuthTransactionView:
+ config = _config()
+ if config.backend != "official" or not config.client_id or not config.client_secret:
+ raise TwitterSetupConflict("Configure the Twitter OAuth App first")
+ transaction_id = str(uuid.uuid4())
+ provider_state = secrets.token_urlsafe(32)
+ verifier, challenge = _pkce_pair()
+ redirect_uri = _redirect_uri()
+ now = _now()
+ transaction = OAuthTransaction(
+ status="pending",
+ provider_state=provider_state,
+ pkce_verifier=verifier,
+ app_fingerprint=_fingerprint(config),
+ redirect_uri=redirect_uri,
+ created_at=now,
+ expires_at=now + TRANSACTION_LIFETIME,
+ )
+
+ def update(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ now = _now()
+ transactions = {
+ key: _terminal(value, "expired", error="Superseded by a newer setup")
+ if value.status in {"pending", "exchanging"}
+ else value
+ for key, value in state.oauth_transactions.items()
+ if value.closed_at is None or value.closed_at + TERMINAL_RETENTION > now
+ }
+ transactions[transaction_id] = transaction
+ ordered = sorted(
+ transactions.items(), key=lambda item: item[1].created_at, reverse=True
+ )[:MAX_TRANSACTIONS]
+ state.oauth_transactions = dict(ordered)
+ return state
+
+ _extension().mutate_state(update)
+ client = OAuth2Client(
+ client_id=config.client_id,
+ client_secret=config.client_secret,
+ scope=" ".join(SCOPES),
+ redirect_uri=redirect_uri,
+ token_endpoint_auth_method="client_secret_basic",
+ timeout=10,
+ )
+ try:
+ authorize_url, _ = client.create_authorization_url(
+ AUTHORIZE_URL,
+ state=provider_state,
+ code_verifier=verifier,
+ code_challenge=challenge,
+ code_challenge_method="S256",
+ )
+ finally:
+ typing.cast(httpx.Client, client).close()
+ return _transaction_view(transaction_id, transaction, authorize_url=authorize_url)
+
+
+def get_oauth_transaction(transaction_id: str) -> OAuthTransactionView:
+ transaction = _state().oauth_transactions.get(transaction_id)
+ if transaction is None:
+ raise TwitterSetupError("OAuth transaction not found")
+ if transaction.status in {"pending", "exchanging"} and transaction.expires_at <= _now():
+
+ def expire(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ current = state.oauth_transactions.get(transaction_id)
+ if current is not None and current.status in {"pending", "exchanging"}:
+ state.oauth_transactions[transaction_id] = _terminal(
+ current, "expired", error="OAuth transaction expired"
+ )
+ return state
+
+ state = typing.cast(TwitterExtensionState, _extension().mutate_state(expire))
+ transaction = state.oauth_transactions[transaction_id]
+ return _transaction_view(transaction_id, transaction)
+
+
+async def _exchange_code(
+ config: TwitterExtensionConfig,
+ transaction: OAuthTransaction,
+ code: str,
+) -> tuple[dict[str, typing.Any], str, str]:
+ client = AsyncOAuth2Client(
+ client_id=config.client_id,
+ client_secret=config.client_secret,
+ redirect_uri=transaction.redirect_uri,
+ token_endpoint_auth_method="client_secret_basic",
+ timeout=10,
+ )
+ try:
+ try:
+ token = await client.fetch_token(
+ TOKEN_URL,
+ code=code,
+ code_verifier=transaction.pkce_verifier,
+ )
+ response = await typing.cast(httpx.AsyncClient, client).get(CURRENT_USER_URL)
+ response.raise_for_status()
+ payload = response.json()
+ if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict):
+ raise TwitterProviderError("Twitter returned an invalid account response")
+ user = payload["data"]
+ user_id = user.get("id")
+ handle = user.get("username")
+ if not isinstance(user_id, str) or not isinstance(handle, str):
+ raise TwitterProviderError("Twitter user response is incomplete")
+ return dict(token), user_id, handle
+ except OAuthError:
+ raise TwitterProviderError("Twitter rejected the authorization exchange") from None
+ except httpx.TimeoutException:
+ raise TwitterProviderError("Twitter authorization timed out") from None
+ except httpx.HTTPError:
+ raise TwitterProviderError("Twitter authorization request failed") from None
+ except (TypeError, ValueError):
+ raise TwitterProviderError(
+ "Twitter returned an invalid authorization response"
+ ) from None
+ finally:
+ await typing.cast(httpx.AsyncClient, client).aclose()
+
+
+def _claim_callback(provider_state: str) -> tuple[str, OAuthTransaction]:
+ box: dict[str, typing.Any] = {}
+
+ def claim(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ matched = next(
+ (
+ (key, value)
+ for key, value in state.oauth_transactions.items()
+ if value.provider_state == provider_state
+ ),
+ None,
+ )
+ if matched is None:
+ raise TwitterSetupConflict("OAuth transaction is unknown")
+ transaction_id, transaction = matched
+ if transaction.status != "pending" or transaction.expires_at <= _now():
+ raise TwitterSetupConflict("OAuth transaction is no longer active")
+ box["id"] = transaction_id
+ box["transaction"] = transaction
+ state.oauth_transactions[transaction_id] = transaction.model_copy(
+ update={"status": "exchanging"}
+ )
+ return state
+
+ _extension().mutate_state(claim)
+ return typing.cast(str, box["id"]), typing.cast(OAuthTransaction, box["transaction"])
+
+
+def _finish_callback(
+ transaction_id: str,
+ transaction: OAuthTransaction,
+ *,
+ account: TwitterAccount | None = None,
+ error: str | None = None,
+) -> None:
+ def finish(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ current = state.oauth_transactions.get(transaction_id)
+ if (
+ current is None
+ or current.status != "exchanging"
+ or current.provider_state != transaction.provider_state
+ or current.app_fingerprint != transaction.app_fingerprint
+ ):
+ raise TwitterSetupConflict("OAuth transaction was superseded")
+ if account is None:
+ state.oauth_transactions[transaction_id] = _terminal(
+ current, "failed", error=error or "Twitter authorization failed"
+ )
+ else:
+ state.account = account
+ state.oauth_transactions[transaction_id] = _terminal(current, "succeeded")
+ return state
+
+ _extension().mutate_state(finish)
+
+
+def _callback_html(status_code: int, title: str, message: str) -> HTMLResponse:
+ safe_title = html.escape(title)
+ safe_message = html.escape(message)
+ return HTMLResponse(
+ ""
+ f"{safe_title}{safe_title}
"
+ f"{safe_message}
You can close this window.
",
+ status_code=status_code,
+ )
+
+
+async def oauth_callback(
+ code: str | None = None,
+ state: str | None = None,
+ error: str | None = None,
+ error_description: str | None = None,
+) -> HTMLResponse:
+ transaction_id: str | None = None
+ transaction: OAuthTransaction | None = None
+ if not state:
+ return _callback_html(400, "Twitter setup failed", "Missing OAuth state.")
+ try:
+ transaction_id, transaction = _claim_callback(state)
+ if error:
+ message = (
+ "Twitter authorization was declined"
+ if error == "access_denied"
+ else "Twitter returned an authorization error"
+ )
+ _finish_callback(transaction_id, transaction, error=message)
+ return _callback_html(400, "Twitter setup declined", message)
+ if not code:
+ raise TwitterSetupConflict("Missing authorization code")
+ config = _config()
+ if _fingerprint(config) != transaction.app_fingerprint:
+ raise TwitterSetupConflict("OAuth App changed during authorization")
+ token, user_id, handle = await _exchange_code(config, transaction, code)
+ raw_scope = token.get("scope", "")
+ scopes = tuple(raw_scope.split()) if isinstance(raw_scope, str) else SCOPES
+ account = TwitterAccount(
+ token=token,
+ user_id=user_id,
+ handle=handle,
+ scopes=scopes,
+ app_fingerprint=transaction.app_fingerprint,
+ authorization_id=str(uuid.uuid4()),
+ connected_at=_now(),
+ )
+ _finish_callback(transaction_id, transaction, account=account)
+ except TwitterSetupConflict as failure:
+ message = _bounded_provider_error(failure)
+ try:
+ if transaction_id is not None and transaction is not None:
+ _finish_callback(transaction_id, transaction, error=message)
+ except TwitterSetupError:
+ pass
+ return _callback_html(400, "Twitter setup failed", message)
+ except (httpx.HTTPError, TwitterSetupError) as failure:
+ message = _bounded_provider_error(failure)
+ try:
+ if transaction_id is not None and transaction is not None:
+ _finish_callback(transaction_id, transaction, error=message)
+ except TwitterSetupError:
+ pass
+ return _callback_html(502, "Twitter is unavailable", message)
+ return _callback_html(200, "Twitter connected", "Authorization completed successfully.")
+
+
+def disconnect_account() -> TwitterSetupStatus:
+ def disconnect(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ state = TwitterExtensionState.model_validate(model.model_dump())
+ state.account = None
+ state.oauth_transactions = {
+ key: _terminal(value, "expired", error="Account disconnected")
+ if value.status in {"pending", "exchanging"}
+ else value
+ for key, value in state.oauth_transactions.items()
+ }
+ return state
+
+ _extension().mutate_state(disconnect)
+ return get_setup_status()
+
+
+def ensure_bookmark_source(body: BookmarkSourceInput) -> TwitterSetupStatus:
+ state = _state()
+ if state.account is None or state.account.reconnect_required:
+ raise TwitterSetupConflict("Connect a Twitter account first")
+ if body.source_id is not None:
+ with SessionLocal() as db:
+ source = db.get(SourceModel, body.source_id)
+ if source is None or source.type != BOOKMARK_SOURCE_TYPE:
+ raise TwitterSetupConflict("Bookmark Source does not exist")
+ else:
+ source, _ = SourceManager.ensure_exists(
+ BOOKMARK_SOURCE_TYPE,
+ nickname=body.nickname,
+ collect_at=body.collect_at,
+ )
+ if source.id is None:
+ raise TwitterSetupError("Bookmark Source has no identifier")
+
+ def select(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ selected = TwitterExtensionState.model_validate(model.model_dump())
+ selected.bookmark_source_id = source.id
+ return selected
+
+ _extension().mutate_state(select)
+ return get_setup_status()
+
+
+async def finish_setup() -> TwitterSetupStatus:
+ from .api import OfficialAPI
+ from app.business.source import SourceCollectJobManager
+
+ state = _state()
+ account = state.account
+ if account is None or account.reconnect_required:
+ raise TwitterSetupConflict("Connect a Twitter account first")
+ if state.bookmark_source_id is None:
+ raise TwitterSetupConflict("Configure a Bookmark Source first")
+ api = OfficialAPI.from_extension(expected_authorization_id=account.authorization_id)
+ try:
+ user_id, handle = await api.get_user()
+ finally:
+ await api.close()
+
+ def update_identity(model: sqlmodel.SQLModel) -> sqlmodel.SQLModel:
+ current = TwitterExtensionState.model_validate(model.model_dump())
+ if (
+ current.account is None
+ or current.account.authorization_id != account.authorization_id
+ ):
+ raise TwitterSetupConflict("Twitter account changed during setup")
+ current.account.user_id = user_id
+ current.account.handle = handle
+ return current
+
+ _extension().mutate_state(update_identity)
+ with SessionLocal() as db:
+ source = db.get(SourceModel, state.bookmark_source_id)
+ if source is None or source.type != BOOKMARK_SOURCE_TYPE:
+ raise TwitterSetupConflict("Bookmark Source no longer exists")
+ SourceCollectJobManager.ensure(
+ state.bookmark_source_id,
+ {
+ "full": False,
+ "result_limit": 40,
+ "authorization_id": account.authorization_id,
+ },
+ )
+ await SourceCollectJobManager.check()
+ return get_setup_status()
+
+
+def _http_error(error: TwitterSetupError) -> typing.NoReturn:
+ status = 404 if str(error) == "OAuth transaction not found" else 409
+ raise fastapi.HTTPException(status_code=status, detail=str(error)) from error
+
+
+def register_setup_routes(router: fastapi.APIRouter) -> None:
+ @router.get("/setup", response_model=TwitterSetupStatus)
+ def status():
+ return get_setup_status()
+
+ @router.put("/setup/oauth-app", response_model=TwitterSetupStatus)
+ def oauth_app(body: OAuthAppInput):
+ try:
+ return save_oauth_app(body)
+ except TwitterSetupError as failure:
+ _http_error(failure)
+
+ @router.post(
+ "/setup/oauth-transactions",
+ response_model=OAuthTransactionView,
+ status_code=fastapi.status.HTTP_201_CREATED,
+ )
+ def start_oauth():
+ try:
+ return begin_oauth()
+ except TwitterSetupError as failure:
+ _http_error(failure)
+
+ @router.get(
+ "/setup/oauth-transactions/{transaction_id}",
+ response_model=OAuthTransactionView,
+ )
+ def transaction(transaction_id: str):
+ try:
+ return get_oauth_transaction(transaction_id)
+ except TwitterSetupError as failure:
+ _http_error(failure)
+
+ @router.delete("/setup/account", response_model=TwitterSetupStatus)
+ def disconnect():
+ return disconnect_account()
+
+ @router.post("/setup/bookmark-source", response_model=TwitterSetupStatus)
+ def bookmark_source(body: BookmarkSourceInput):
+ try:
+ return ensure_bookmark_source(body)
+ except TwitterSetupError as failure:
+ _http_error(failure)
+
+ @router.post("/setup/finish", response_model=TwitterSetupStatus)
+ async def finish():
+ try:
+ return await finish_setup()
+ except TwitterSetupError as failure:
+ _http_error(failure)
+
+ router.add_api_route(
+ "/auth/callback",
+ oauth_callback,
+ methods=["GET"],
+ response_class=HTMLResponse,
+ )
diff --git a/migrations/revision-integrity.json b/migrations/revision-integrity.json
index ea13806..4a49a04 100644
--- a/migrations/revision-integrity.json
+++ b/migrations/revision-integrity.json
@@ -6,6 +6,7 @@
"a1b2c3d4e5f6_add_clients_modify_extensions_enabled.py": "15292a9fcf834474c8453e3d1d9db962962fb5d729bc0b80053615cf71315b96",
"b8c1d2e3f4a5_native_extension_distribution_cutover.py": "c0527edff210a9646c69de2da01c2877bce4acbc05af969c5bfb8d1a3fd0d3c8",
"c4e8a7b6d5f0_converge_production_schema.py": "fb5687ad523c64297fe101d109918af2501bec26219de78353944a762eadfe4e",
+ "c7d8e9f0a1b2_add_extension_state.py": "f59486066581eef1ae65d07128a6c5bbe652b959b2004902e81001e629a011bf",
"d9f4e2a1b7c3_adopt_peer_database_protocol.py": "9a163533b5a0619e51bfb94ba7ad0c5c3d168fba5808b08f67531fd8f7e5f263",
"e5a01f9e69ef_init.py": "a10de6c0818abc593951a34e51a974ffcd999f60ba5afc441f87ac6cef8d869e",
"f2a6c8e4b1d7_add_extension_registry_deployment_state.py": "a2cf3059f87e17292baf9fd10a1acb8b26ccb928845dbe4bec95b374164cd46d"
diff --git a/migrations/versions/c7d8e9f0a1b2_add_extension_state.py b/migrations/versions/c7d8e9f0a1b2_add_extension_state.py
new file mode 100644
index 0000000..2535f47
--- /dev/null
+++ b/migrations/versions/c7d8e9f0a1b2_add_extension_state.py
@@ -0,0 +1,103 @@
+"""Add deployment-wide Extension state authority.
+
+Revision ID: c7d8e9f0a1b2
+Revises: b8c1d2e3f4a5
+Create Date: 2026-08-13
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "c7d8e9f0a1b2"
+down_revision: str | Sequence[str] | None = "b8c1d2e3f4a5"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_PROTOCOL_SCHEMA = "inkcre"
+_INTERNAL_SCHEMA = "inkcre_internal"
+
+
+def _replace_state_guard(*, include_extension_state: bool) -> None:
+ state_insert_guard = """
+ IF TG_OP = 'INSERT' AND NEW.state <> '{}'::jsonb THEN
+ RAISE EXCEPTION 'extensions.state must be empty on insert'
+ USING ERRCODE = '23514';
+ END IF;
+ """ if include_extension_state else ""
+ state_version_guard = """
+ IF TG_OP = 'UPDATE'
+ AND OLD.state <> '{}'::jsonb
+ AND NEW.version IS DISTINCT FROM OLD.version
+ THEN
+ RAISE EXCEPTION 'cannot change extension version while state is not empty'
+ USING ERRCODE = '23514';
+ END IF;
+ """ if include_extension_state else ""
+ op.execute(
+ f"""
+ CREATE OR REPLACE FUNCTION {_INTERNAL_SCHEMA}.enforce_extension_state_authority()
+ RETURNS trigger
+ LANGUAGE plpgsql
+ SET search_path = pg_catalog
+ AS $$
+ BEGIN
+ IF TG_OP = 'INSERT' AND cardinality(NEW.enabled) <> 0 THEN
+ RAISE EXCEPTION 'extensions.enabled must be empty on insert'
+ USING ERRCODE = '23514';
+ END IF;
+ {state_insert_guard}
+ IF TG_OP = 'UPDATE'
+ AND cardinality(OLD.enabled) <> 0
+ AND NEW.version IS DISTINCT FROM OLD.version
+ THEN
+ RAISE EXCEPTION 'cannot change extension version while peers are enabled'
+ USING ERRCODE = '23514';
+ END IF;
+ {state_version_guard}
+ IF TG_OP = 'DELETE' AND cardinality(OLD.enabled) <> 0 THEN
+ RAISE EXCEPTION 'cannot delete extension while peers are enabled'
+ USING ERRCODE = '23514';
+ END IF;
+ IF TG_OP = 'DELETE' THEN
+ RETURN OLD;
+ END IF;
+ RETURN NEW;
+ END
+ $$
+ """
+ )
+
+
+def upgrade() -> None:
+ op.add_column(
+ "extensions",
+ sa.Column(
+ "state",
+ postgresql.JSONB(astext_type=sa.Text()),
+ server_default=sa.text("'{}'::jsonb"),
+ nullable=False,
+ ),
+ schema=_PROTOCOL_SCHEMA,
+ )
+ op.create_check_constraint(
+ "extensions_state_object",
+ "extensions",
+ "jsonb_typeof(state) = 'object'",
+ schema=_PROTOCOL_SCHEMA,
+ )
+ _replace_state_guard(include_extension_state=True)
+
+
+def downgrade() -> None:
+ _replace_state_guard(include_extension_state=False)
+ op.drop_constraint(
+ "extensions_state_object",
+ "extensions",
+ schema=_PROTOCOL_SCHEMA,
+ type_="check",
+ )
+ op.drop_column("extensions", "state", schema=_PROTOCOL_SCHEMA)
diff --git a/pdm.lock b/pdm.lock
index 1a9c6e2..0a4a45b 100644
--- a/pdm.lock
+++ b/pdm.lock
@@ -5,7 +5,7 @@
groups = ["default", "dev", "extension-publisher"]
strategy = ["inherit_metadata"]
lock_version = "4.5.0"
-content_hash = "sha256:d1b8b349aba7c33a66d27f83d0b8443fef44ec213bd6a3896009aa553b22dbbb"
+content_hash = "sha256:b98b0fc3750aefd771f27b2cda0e677bb72d46d779d76b5ea9a8eb243c5a2058"
[[metadata.targets]]
requires_python = ">=3.12,<3.13"
@@ -189,6 +189,21 @@ files = [
{file = "attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32"},
]
+[[package]]
+name = "authlib"
+version = "1.7.2"
+requires_python = ">=3.10"
+summary = "The ultimate Python library in building OAuth and OpenID Connect servers and clients."
+groups = ["default"]
+dependencies = [
+ "cryptography",
+ "joserfc>=1.6.0",
+]
+files = [
+ {file = "authlib-1.7.2-py2.py3-none-any.whl", hash = "sha256:3e1faedc9d87e7d56a164eca3ccb6ace0d61b94abe83e92242f8dc8bba9b4a9f"},
+ {file = "authlib-1.7.2.tar.gz", hash = "sha256:2cea25fefcd4e7173bdf1372c0afc265c8034b23a8cd5dcb6a9164b826c64231"},
+]
+
[[package]]
name = "beautifulsoup4"
version = "4.15.0"
@@ -289,26 +304,45 @@ files = [
[[package]]
name = "charset-normalizer"
-version = "3.4.9"
+version = "3.5.0"
requires_python = ">=3.7"
summary = "The Real First Universal Charset Detector. Open, modern and actively maintained alternative to Chardet."
groups = ["default"]
files = [
- {file = "charset_normalizer-3.4.9-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:45b0cc4e3556cd875e09102988d1ab8356c998b596c9fced84547c8138b487a0"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9b2aff1c7b3884512b9512c3eaadd9bab39fb45042ffaaa1dd08ff2b9f8109d9"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:9104ed0bd76a429d46f9ec0dbc9b08ad1d2dcdf2b00a5a0daa1c145329b35b44"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:7b86a2b16095d250c6f58b3d9b2eee6f4147754344f3dab0922f7c9bf7d226c9"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5e226f6218febc71f6c1fc2fafb91c226f75bdc1d8fb12d66823716e891608fd"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-manylinux_2_31_armv7l.whl", hash = "sha256:90c44bc373b7687f6948b693cceaea1348ae0975d7474746559494468e3c1d84"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:9cdef90ae47919cae358d8ab15797a800ed41da7aba5d72419fb510729e2ed4b"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:60f44ade2cf573dad7a277e6f8ca9a51a21dda572b13bd7d8539bb3cd5dbedde"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-musllinux_1_2_armv7l.whl", hash = "sha256:a1786910334ed46ab1dd73222f2cd1e05c2c3bb39f6dddb4f8b36fc382058a39"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:03d07803992c6c7bbc976327f34b18b6160327fc81cb82c9d504720ac0be3b62"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-win32.whl", hash = "sha256:78841cccf1af7b40f6f716338d50c0902dbe88d9f800b3c973b7a9a0a693a642"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-win_amd64.whl", hash = "sha256:4b3dac63058cc36820b0dd072f89898604e2d39686fe05321729d00d8ac185a0"},
- {file = "charset_normalizer-3.4.9-cp312-cp312-win_arm64.whl", hash = "sha256:78fa18e436a1a0e58dbd7e02fc4473f3f32cceb12df9dfca542d075961c307d2"},
- {file = "charset_normalizer-3.4.9-py3-none-any.whl", hash = "sha256:68e5f26a1ad57ded6d1cfb85331d1c1a195314756471d97758c48498bb4dcdf5"},
- {file = "charset_normalizer-3.4.9.tar.gz", hash = "sha256:673611bbd43f0810bec0b0f028ddeaaa501190339cac411f347ac76917c3ae7b"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:98820e1ceb25c6df7a80c4fd8efa59cb121f99bc7c4c1693ad94a2caff5b311d"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:608553f476fca509537e804c4a71f5eb166ce63b75141f89c2c686ce1aa36956"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:6753de11eef42f1c321b26d682957d92c7f7bbce6530f34bbe0f9291dd37cc6f"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:0f76dc0a47f94cb9b69d86f01e477f4b0371ca70208b9ccea7e063c41eed9046"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:c387c6bf91b4774e359a48a179e2872b8e8bf741e4fde06ba8d1665eb9a4760a"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:14f6904a3cf870abf044df3a8c4924ac6c8ef77e9896586fd37e73ae96cff2af"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:0cce46dd29d73e135e8087b96eb62a4aca6d69391b7f97808c6588ebed3178f3"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:b476cdb63df22da2b91837593380be3ddbe406f36c506c1c91d80e7196b66288"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-musllinux_1_2_armv7l.whl", hash = "sha256:1f56ce84b317ef2a59d7d3461891c7597c79247d2192bb8114c68a1a1debfcc0"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:9ce0f885239357379d92fd9a5fddbe20f0e30e0527c29ba69f8e99eeb1304a76"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-musllinux_1_2_riscv64.whl", hash = "sha256:96ae7ab5d8155fde927aa0864fbc8ba3cc4fde6d41ab0c7cea9d6012b4978603"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:bf91921009025e96ce57a03ced6d14604fc3baf0530351638e9504a55da6fa3b"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:0b2e44e6d42d1a4ff78ccc219a93c5449105d10b16198d1aea581080df8073f9"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-win32.whl", hash = "sha256:deb99535e9bf0bea8e274c6413eb939a21be35a3f492678dba4d5b1f4d70f142"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-win_amd64.whl", hash = "sha256:e54dd1a66fa4bce0ccaf0db9dde336e49b3eec646dc4c1c0991279369d373a14"},
+ {file = "charset_normalizer-3.5.0-cp312-cp312-win_arm64.whl", hash = "sha256:b8ea208b304587d47931b36481342d20336e0d338ab052f8b4305926482598d6"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-macosx_10_9_universal2.whl", hash = "sha256:5a4ee37248dfac25107c758bda99d545ce73e60b44d2dd39e4a2bb9f2831e9f5"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:a864bdcacd8bff58bb4845304e031f821a3ec64b2b7259f2d409cd49c9e59ca3"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:84b736e3b391601bc47b86da381c749c0f894e9191aaca9f31f30c2632206df3"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:6abb1f356fb865baeb6ebc3fadd843e9a96fbf49b9adcca55037f3cceccb7438"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:1d366548d2ee28a8cfdcc4296363978cc644a728333be9824d2de4652e83df0a"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:d672f329ae504ee240eb39b6effb3318aa8e7e8924c0ce8eee5760b3fad98539"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:c54036a518748b6c02e666f6d46c3817561998fb904c3be25b56fb4fe3dc5706"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:22a1889f1c9b752c63c36758a0c2145458e3cadb20fced7a0790002e9dd12b26"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:b7eb3eab5c646d3de7dcb14a7c9caebace5249c5767da39e1761cb1576e521a3"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-musllinux_1_2_ppc64le.whl", hash = "sha256:2080aa129a28267984cdc902898993d788c995c384e285d0d19199f56760d52e"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:fd68c825548a611158230e2f9222e210ceb2e3391995c0aa5865cbdf3ab4bd49"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-musllinux_1_2_s390x.whl", hash = "sha256:d8a9316f4da85e937242642b537c6d55d7e9287dd38e5634732f8233932aff45"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:d90254c8f609338c53ec180fcd4c4f9c16502e238e3fc88ca7fd4c2f38d445b8"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-win32.whl", hash = "sha256:8b3e9e29b8b07cc461b9ce7768db7693a93979d0dadf22046f6f3555ded2f516"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-win_amd64.whl", hash = "sha256:0c8953d9d1617794cfc40d81179571c9ba3805dd029623a15c93f1fb70e60a74"},
+ {file = "charset_normalizer-3.5.0-cp37-abi3-win_arm64.whl", hash = "sha256:562d24ca7797c1af8852994950c2e623a907b201fc4b0ed29e92af173d3828ca"},
+ {file = "charset_normalizer-3.5.0-py3-none-any.whl", hash = "sha256:993dfcbe75a85a3784abb5084f2c41b915767c90546fcc92803cffa28611baea"},
+ {file = "charset_normalizer-3.5.0.tar.gz", hash = "sha256:49bd5feb59b0bf3cbf6ebcf4352e371c95b9da9bacd4449f8b64d0ad2c10a26e"},
]
[[package]]
@@ -692,6 +726,20 @@ files = [
{file = "jiter-0.16.0.tar.gz", hash = "sha256:7b24c3492c5f4f84a37946ad9cf504910cf6a782d6a4e0689b6673c5894b4a1c"},
]
+[[package]]
+name = "joserfc"
+version = "1.7.4"
+requires_python = ">=3.10"
+summary = "The ultimate Python library for JOSE RFCs, including JWS, JWE, JWK, JWA, JWT"
+groups = ["default"]
+dependencies = [
+ "cryptography>=45.0.1",
+]
+files = [
+ {file = "joserfc-1.7.4-py3-none-any.whl", hash = "sha256:32d46c2cd5e3203c13e87a6c61333cab310b1ba80cd54b4c4f386a848a122463"},
+ {file = "joserfc-1.7.4.tar.gz", hash = "sha256:b3bc561672ae541b17a9237053b48a03dacddd92d68047b3ecdfb4b5714a88ed"},
+]
+
[[package]]
name = "js2py-3-13"
version = "0.74.1"
@@ -1363,7 +1411,7 @@ files = [
[[package]]
name = "python-discovery"
-version = "1.5.1"
+version = "1.5.2"
requires_python = ">=3.8"
summary = "Python interpreter discovery"
groups = ["dev"]
@@ -1371,8 +1419,8 @@ dependencies = [
"filelock>=3.15.4",
]
files = [
- {file = "python_discovery-1.5.1-py3-none-any.whl", hash = "sha256:ac07f44cade589d954e9d6a1e1468539fdddd2cf676beb51da73e0f156b7c932"},
- {file = "python_discovery-1.5.1.tar.gz", hash = "sha256:e2ea8b884cd1701f386eda8cf327b87743f1dc21b7f784470799537d95635384"},
+ {file = "python_discovery-1.5.2-py3-none-any.whl", hash = "sha256:3e338c2d0f15dfaeea57493f4c2c6caebe0e998ea815c30ae8bf8ee21f1112d3"},
+ {file = "python_discovery-1.5.2.tar.gz", hash = "sha256:45fd4f20a4e3f9b7bf2e0817870bc8e3b320a19658da177af800768c82dbf354"},
]
[[package]]
@@ -1549,7 +1597,7 @@ files = [
[[package]]
name = "sqlalchemy"
-version = "2.0.51"
+version = "2.0.52"
requires_python = ">=3.7"
summary = "Database Abstraction Library"
groups = ["default"]
@@ -1559,15 +1607,15 @@ dependencies = [
"typing-extensions>=4.6.0",
]
files = [
- {file = "sqlalchemy-2.0.51-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7d78702b26ba1c18b2d0fb2ea940ba7f17a9581b42e8361ff93920ebbee1235a"},
- {file = "sqlalchemy-2.0.51-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:581921d849d6e6f994d560389192955e80e2950e18fcdfe2ccea863e01158e6e"},
- {file = "sqlalchemy-2.0.51-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1d21ce524ab86c23046e992a5b81cb54c21079c6df6e78b8fc77d77cac70a6b9"},
- {file = "sqlalchemy-2.0.51-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:c5d98a2709840027f5a347c3af0a7c3d5f6c1ff93af2ca1c54494e23cba8f389"},
- {file = "sqlalchemy-2.0.51-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:1181256e0f16479691b5616d36375dc2620ad8332b25978763c3d206ad3f3f1d"},
- {file = "sqlalchemy-2.0.51-cp312-cp312-win32.whl", hash = "sha256:9f380393be5abeb6815f68fd39271b95127173511b6706b0a630a9995d53f8f5"},
- {file = "sqlalchemy-2.0.51-cp312-cp312-win_amd64.whl", hash = "sha256:2cf39aabdf48e87c1c2c2ed6d20d33ffa0733b3071ce9c5f66357947dd009080"},
- {file = "sqlalchemy-2.0.51-py3-none-any.whl", hash = "sha256:bb024d8b621d0be75f4f44ecc7c950450026e76d66dc8f791bb5331d7fed59d5"},
- {file = "sqlalchemy-2.0.51.tar.gz", hash = "sha256:804dccd8a4a6242c4e30ad961e540e18a588f6527202f2d6791b01845d59fdc9"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:be8c49131665dfe2cc74c498aa1240ffb548d0fd901325dd11c2c7a18956f727"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1b2d9e507a458832adcfbd8af6e2036ddf069b7710b799448542ebccae2dceee"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8738008376d22f30f411ea3efecf39b51110b6996d80bb73786f30bcfdd5fd3b"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:37a4d548327b6cab9c7d8cdb4e0e82feabee0110c4d150059068e2d1cfbd99ee"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:e49f51a5d59857a7a0dcaf9469febf7197d9394bd88f00d69c2c4e848112cdbf"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-win32.whl", hash = "sha256:afda3ec521d0517d0de783fc70030775841900896d832de5bbd066549290470e"},
+ {file = "sqlalchemy-2.0.52-cp312-cp312-win_amd64.whl", hash = "sha256:2d5e53e36e37129fe0be8b9d08b6e4052c10a963ee6cda56c8c10dcc194b99ca"},
+ {file = "sqlalchemy-2.0.52-py3-none-any.whl", hash = "sha256:3b81b8363a919ce53453591cdb93702e6bd54ade6c4fa2f468fc053baee5ed89"},
+ {file = "sqlalchemy-2.0.52.tar.gz", hash = "sha256:5e2d46356ac2ccb7d268ab6c2319ac6a2b42f1b8d5fd8bd3d46855cd82abee97"},
]
[[package]]
@@ -1602,15 +1650,15 @@ files = [
[[package]]
name = "tencentcloud-sdk-python-common"
-version = "3.1.154"
+version = "3.1.155"
summary = "Tencent Cloud Common SDK for Python"
groups = ["default"]
dependencies = [
"requests>=2.16.0",
]
files = [
- {file = "tencentcloud_sdk_python_common-3.1.154-py2.py3-none-any.whl", hash = "sha256:47d58eebca6eb654f7c65af9abb94769b930ba118dc3e9703db8a32980bd45f1"},
- {file = "tencentcloud_sdk_python_common-3.1.154.tar.gz", hash = "sha256:3c8ef9895bb0180c4f435b96cf61612df467de4b6c8bb5f77fb102aa50f16ade"},
+ {file = "tencentcloud_sdk_python_common-3.1.155-py2.py3-none-any.whl", hash = "sha256:20549fd7147f70f9345b427033f7873728edf491c8e86ea34a61667bb5ceca87"},
+ {file = "tencentcloud_sdk_python_common-3.1.155.tar.gz", hash = "sha256:3c8c3c15530e8464c62ac44d632a4c95399df5d546e29d9d45fdad809b98618b"},
]
[[package]]
@@ -1686,7 +1734,7 @@ files = [
[[package]]
name = "typing-inspection"
-version = "0.4.3"
+version = "0.4.4"
requires_python = ">=3.10"
summary = "Runtime typing introspection tools"
groups = ["default"]
@@ -1694,8 +1742,8 @@ dependencies = [
"typing-extensions>=4.15.0",
]
files = [
- {file = "typing_inspection-0.4.3-py3-none-any.whl", hash = "sha256:5f42b23858a91e0b4ef521f5418f03a0da3c9216fd2995ef5e73463100e676cd"},
- {file = "typing_inspection-0.4.3.tar.gz", hash = "sha256:c5f9ec1530b5c1e2c9bc34a84d9a3466ed1b2f3f2fa9f901368d9c5596210e4d"},
+ {file = "typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147"},
+ {file = "typing_inspection-0.4.4.tar.gz", hash = "sha256:547274fa6b0a561ccf549cc9524b999a578e737d015d8709d021f9d0d13bea47"},
]
[[package]]
diff --git a/pyproject.toml b/pyproject.toml
index f8dcb0d..88eacb3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,11 +1,12 @@
[project]
name = "inkcre-core"
-version = "0.1.0"
+version = "0.1.1"
description = "Backend of InKCre."
authors = [{ name = "Lan_zhijiang", email = "lanzhijiang@foxmail.com" }]
readme = "README.md"
requires-python = ">=3.12,<3.13"
-dependencies = [
+dependencies = [
+ "authlib>=1.7.2,<2.0.0",
"pydantic (>=2.11.7,<3.0.0)",
"pydantic-settings (>=2.14.2,<3.0.0)",
"fastapi (>=0.139.2,<0.140.0)",
@@ -29,7 +30,8 @@ dependencies = [
"requests (>=2.33.0,<3.0.0)",
"sqlmodel (>=0.0.24,<0.0.25)",
"apscheduler>=3.11.0",
- "html2text>=2025.4.15",
+ "html2text>=2025.4.15",
+ "httpx>=0.28.1,<0.29.0",
"logtail-python>=0.3.0",
"alembic>=1.16.4",
"alembic-utils>=0.8.8",
diff --git a/scripts/dev_database.py b/scripts/dev_database.py
index 0f82054..274982b 100644
--- a/scripts/dev_database.py
+++ b/scripts/dev_database.py
@@ -300,6 +300,8 @@ def _readiness(state: Mapping[str, Any]) -> dict[str, Any]:
"--rm",
"--no-deps",
"init",
+ "python",
+ "scripts/container.py",
"db",
"ready",
"--profile",
@@ -450,6 +452,8 @@ def reset(instance: str, confirmed: bool) -> dict[str, Any]:
"--rm",
"--no-deps",
"init",
+ "python",
+ "scripts/container.py",
"db",
"reset-dev",
"--confirm",
diff --git a/scripts/dev_database_provider.py b/scripts/dev_database_provider.py
index 60df939..280d8eb 100644
--- a/scripts/dev_database_provider.py
+++ b/scripts/dev_database_provider.py
@@ -31,6 +31,7 @@
Path("migrations"),
Path("pdm.lock"),
Path("pyproject.toml"),
+ Path("release/database-contract"),
Path("run.py"),
Path("scripts"),
Path("utils"),
@@ -398,7 +399,10 @@ def diagnose_database_provider(provider: DatabaseProvider) -> ProviderDiagnostic
def control_socket(identity: str) -> Path:
"""Return the instance-owned OpenSSH control socket path."""
- return RUNTIME_ROOT / "ssh" / identity
+ # OpenSSH control sockets use a short Unix-domain path limit. Worktree paths can
+ # exceed it before the instance suffix is added, so keep only the stable owner
+ # and instance identity in the system temporary directory.
+ return Path(tempfile.gettempdir()) / "inkcre-core-py-ssh" / identity
def database_access_ready(
diff --git a/tests/extensions/test_twitter.py b/tests/extensions/test_twitter.py
index 041303a..9028c70 100644
--- a/tests/extensions/test_twitter.py
+++ b/tests/extensions/test_twitter.py
@@ -3,14 +3,25 @@
from pathlib import Path
import subprocess
import sys
+import typing
from urllib.parse import parse_qs, urlparse
+from fastapi.routing import APIRoute
+import httpx
import pytest
from app.business.extension.runtime import ExtensionRuntimeRecord
-from app.business.source import SourceManager
+from app.settings import settings
from extensions.twitter import Extension, TwitterExtensionConfig
-from extensions.twitter.api import OfficialAPI, TwikitAPI, TwitterAPI
+from extensions.twitter.api import TwikitAPI, TwitterAPI
+import extensions.twitter.setup_flow as setup
+from extensions.twitter.api import OfficialAPI
+from extensions.twitter.bookmark import Source as BookmarkSource
+from extensions.twitter.setup_flow import (
+ TwitterAccount,
+ TwitterExtensionState,
+ TwitterSetupConflict,
+)
COOKIE_CONVERTER = (
@@ -22,81 +33,377 @@
)
-def test_get_oauth_authorize_url(monkeypatch):
- monkeypatch.setenv("API_BASE_URL", "https://preview.example")
- api = OfficialAPI(client_id="test-client", client_secret="test-secret")
+@pytest.fixture(autouse=True)
+def clean_twitter_runtime():
+ TwitterAPI.SINGLETON = None
+ Extension.release_runtime()
+ yield
+ TwitterAPI.SINGLETON = None
+ Extension.release_runtime()
+
+
+def attach_runtime(
+ config: TwitterExtensionConfig | None = None,
+ state: TwitterExtensionState | None = None,
+):
+ config_box = (config or TwitterExtensionConfig()).model_dump(mode="json")
+ state_box = (state or TwitterExtensionState()).model_dump(mode="json")
+
+ def read_config():
+ return dict(config_box)
+
+ def persist_config(value):
+ config_box.clear()
+ config_box.update(value)
+
+ def read_state():
+ return dict(state_box)
+
+ def mutate_state(transform):
+ updated = transform(dict(state_box))
+ state_box.clear()
+ state_box.update(updated)
+ return dict(state_box)
+
+ def mutate_config_and_state(transform):
+ updated_config, updated_state = transform(dict(config_box), dict(state_box))
+ config_box.clear()
+ config_box.update(updated_config)
+ state_box.clear()
+ state_box.update(updated_state)
+ return dict(config_box), dict(state_box)
+
+ record = ExtensionRuntimeRecord(
+ extension_id="twitter",
+ config=dict(config_box),
+ read_config=read_config,
+ persist_config=persist_config,
+ read_state=read_state,
+ mutate_state=mutate_state,
+ mutate_config_and_state=mutate_config_and_state,
+ persist_config_schema=lambda schema: None,
+ )
+ setattr(Extension, "__runtime_record__", record)
+ setattr(Extension, "config", TwitterExtensionConfig.model_validate(config_box))
+ return config_box, state_box
+
+
+def test_begin_oauth_uses_pkce_s256_and_the_minimal_bookmark_scopes(monkeypatch):
+ monkeypatch.setattr(settings, "client_base_url", "https://core.example")
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret")
+ )
- parsed = urlparse(api.get_oauth_authorize_url())
+ transaction = setup.begin_oauth()
+ parsed = urlparse(typing.cast(str, transaction.authorize_url))
query = parse_qs(parsed.query)
- assert parsed.scheme == "https"
- assert parsed.netloc == "x.com"
+ assert (parsed.scheme, parsed.netloc, parsed.path) == (
+ "https",
+ "x.com",
+ "/i/oauth2/authorize",
+ )
assert query["client_id"] == ["test-client"]
- assert query["redirect_uri"] == ["https://preview.example/twitter/auth/callback"]
- assert query["code_challenge_method"] == ["plain"]
+ assert query["redirect_uri"] == ["https://core.example/twitter/auth/callback"]
+ assert query["code_challenge_method"] == ["S256"]
+ assert set(query["scope"][0].split()) == set(setup.SCOPES)
+ assert "bookmark.write" not in query["scope"][0]
+ stored = TwitterExtensionState.model_validate(state)
+ assert stored.oauth_transactions[transaction.id].pkce_verifier
-def test_cookie_converter_reads_sensitive_input_from_stdin():
- result = subprocess.run( # noqa: S603
- [sys.executable, str(COOKIE_CONVERTER)],
- input="session=secret; preference=compact",
- check=False,
- capture_output=True,
- text=True,
+def test_new_oauth_supersedes_and_scrubs_the_previous_transaction(monkeypatch):
+ monkeypatch.setattr(settings, "client_base_url", "https://core.example")
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret")
)
- assert result.returncode == 0
- assert json.loads(result.stdout) == {
- "session": "secret",
- "preference": "compact",
- }
+ first = setup.begin_oauth()
+ setup.begin_oauth()
+ prior = TwitterExtensionState.model_validate(state).oauth_transactions[first.id]
+ assert prior.status == "expired"
+ assert prior.provider_state is None
+ assert prior.pkce_verifier is None
-def test_bookmark_api_creates_the_registered_source_type(monkeypatch):
- import fastapi
- from fastapi.routing import APIRoute
- created: list[tuple[str, str | None]] = []
- monkeypatch.setattr(TwitterAPI, "new", lambda api_router=None: object())
+def test_replacing_oauth_app_requires_explicit_account_reset_confirmation(monkeypatch):
+ monkeypatch.setattr(settings, "client_base_url", "https://core.example")
monkeypatch.setattr(
- SourceManager,
- "create",
- lambda source_type, nickname=None: created.append((source_type, nickname)),
+ setup,
+ "_source_and_job_status",
+ lambda state: (state.bookmark_source_id, (), False),
)
- router = fastapi.APIRouter()
- Extension._register_apis(router)
- route = next(
- route
- for route in router.routes
- if isinstance(route, APIRoute) and route.path == "/bookmark"
+ config, state = attach_runtime(
+ TwitterExtensionConfig(client_id="first-client", client_secret="first-secret"),
+ connected_state(),
)
- route.endpoint("Reading")
- assert created == [("extensions.twitter.bookmark.Source", "Reading")]
+ with pytest.raises(TwitterSetupConflict, match="requires confirmation"):
+ setup.save_oauth_app(
+ setup.OAuthAppInput(client_id="next-client", client_secret="next-secret")
+ )
+ assert config["client_id"] == "first-client"
+ assert TwitterExtensionState.model_validate(state).account is not None
-def test_on_close_persists_config_even_when_api_cleanup_fails(monkeypatch):
- persisted: list[dict[str, object]] = []
- record = ExtensionRuntimeRecord(
- extension_id="twitter",
- config={},
- persist_config=persisted.append,
- persist_config_schema=lambda schema: None,
+ status = setup.save_oauth_app(
+ setup.OAuthAppInput(
+ client_id="next-client",
+ client_secret="next-secret",
+ confirm_account_reset=True,
+ )
)
- setattr(Extension, "__runtime_record__", record)
- setattr(Extension, "config", TwitterExtensionConfig(api_language="zh-CN"))
+
+ assert config["client_id"] == "next-client"
+ assert TwitterExtensionState.model_validate(state).account is None
+ assert status.callback_url == "https://core.example/twitter/auth/callback"
+
+
+def test_callback_persists_account_and_scrubs_provider_transaction(monkeypatch):
+ monkeypatch.setattr(settings, "client_base_url", "https://core.example")
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret")
+ )
+ transaction = setup.begin_oauth()
+ stored = TwitterExtensionState.model_validate(state).oauth_transactions[transaction.id]
+
+ async def exchange(config, claimed, code):
+ assert code == "authorization-code"
+ assert claimed.provider_state == stored.provider_state
+ return {"access_token": "secret", "scope": "tweet.read users.read"}, "42", "inkcre"
+
+ monkeypatch.setattr(setup, "_exchange_code", exchange)
+ response = asyncio.run(
+ setup.oauth_callback(code="authorization-code", state=stored.provider_state)
+ )
+ finished = TwitterExtensionState.model_validate(state)
+
+ assert response.status_code == 200
+ assert finished.account is not None
+ assert finished.account.user_id == "42"
+ assert finished.account.authorization_id
+ terminal = finished.oauth_transactions[transaction.id]
+ assert terminal.status == "succeeded"
+ assert terminal.provider_state is None
+ assert terminal.pkce_verifier is None
+
+
+def test_claimed_callback_validation_failure_becomes_terminal(monkeypatch):
+ monkeypatch.setattr(settings, "client_base_url", "https://core.example")
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret")
+ )
+ transaction = setup.begin_oauth()
+ stored = TwitterExtensionState.model_validate(state).oauth_transactions[transaction.id]
+
+ response = asyncio.run(setup.oauth_callback(state=stored.provider_state))
+ terminal = TwitterExtensionState.model_validate(state).oauth_transactions[transaction.id]
+
+ assert response.status_code == 400
+ assert terminal.status == "failed"
+ assert terminal.provider_state is None
+ assert terminal.pkce_verifier is None
+
+
+def test_provider_denial_does_not_reflect_the_callback_description(monkeypatch):
+ monkeypatch.setattr(settings, "client_base_url", "https://core.example")
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret")
+ )
+ transaction = setup.begin_oauth()
+ stored = TwitterExtensionState.model_validate(state).oauth_transactions[transaction.id]
+
+ response = asyncio.run(
+ setup.oauth_callback(
+ state=stored.provider_state,
+ error="access_denied",
+ error_description="provider body that must not be reflected",
+ )
+ )
+ terminal = TwitterExtensionState.model_validate(state).oauth_transactions[transaction.id]
+
+ assert response.status_code == 400
+ assert "provider body" not in bytes(response.body).decode()
+ assert terminal.error == "Twitter authorization was declined"
+
+
+def connected_state(authorization_id: str = "authorization-1") -> TwitterExtensionState:
+ config = TwitterExtensionConfig(client_id="test-client", client_secret="test-secret")
+ return TwitterExtensionState(
+ account=TwitterAccount(
+ token={"access_token": "secret"},
+ user_id="42",
+ handle="inkcre",
+ scopes=("bookmark.read",),
+ app_fingerprint=setup._fingerprint(config),
+ authorization_id=authorization_id,
+ connected_at=setup._now(),
+ )
+ )
+
+
+def test_provider_access_fails_before_network_when_authorization_changed(monkeypatch):
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret"),
+ connected_state(),
+ )
+ api = OfficialAPI.from_extension(expected_authorization_id="authorization-1")
+ state.clear()
+ state.update(TwitterExtensionState().model_dump(mode="json"))
+
+ monkeypatch.setattr(
+ "extensions.twitter.api.AsyncOAuth2Client",
+ lambda *args, **kwargs: pytest.fail("provider client must not be constructed"),
+ )
+ with pytest.raises(TwitterSetupConflict, match="changed before provider access"):
+ asyncio.run(api.get_user())
+
+
+def test_stale_refresh_cannot_replace_a_newer_token(monkeypatch):
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret"),
+ connected_state(),
+ )
+ api = OfficialAPI.from_extension(expected_authorization_id="authorization-1")
+
+ class RefreshingClient:
+ def __init__(self, *args, update_token, **kwargs):
+ self.update_token = update_token
+
+ async def request(self, *args, **kwargs):
+ state["account"]["token"] = {"access_token": "newer-token"}
+ await self.update_token({"access_token": "stale-token"})
+ raise AssertionError("stale refresh must fail before provider response handling")
+
+ async def aclose(self):
+ return None
+
+ monkeypatch.setattr("extensions.twitter.api.AsyncOAuth2Client", RefreshingClient)
+
+ with pytest.raises(TwitterSetupConflict, match="token changed during refresh"):
+ asyncio.run(api._request("GET", "/users/me"))
+ assert state["account"]["token"] == {"access_token": "newer-token"}
+
+
+def test_current_provider_unauthorized_response_requires_reconnect(monkeypatch):
+ _, state = attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret"),
+ connected_state(),
+ )
+ api = OfficialAPI.from_extension(expected_authorization_id="authorization-1")
+
+ class UnauthorizedClient:
+ def __init__(self, *args, **kwargs):
+ return None
+
+ async def request(self, method, url, **kwargs):
+ return httpx.Response(401, request=httpx.Request(method, url))
+
+ async def aclose(self):
+ return None
+
+ monkeypatch.setattr("extensions.twitter.api.AsyncOAuth2Client", UnauthorizedClient)
+
+ with pytest.raises(RuntimeError, match="requires reconnection"):
+ asyncio.run(api._request("GET", "/users/me"))
+ assert state["account"]["reconnect_required"] is True
+
+
+def test_scheduled_bookmark_job_carries_the_current_authorization_identity():
+ attach_runtime(
+ TwitterExtensionConfig(client_id="test-client", client_secret="test-secret"),
+ connected_state("authorization-current"),
+ )
+
+ assert BookmarkSource(_id=1).scheduled_collect_config() == {
+ "full": False,
+ "result_limit": 40,
+ "authorization_id": "authorization-current",
+ }
+
+
+def test_setup_projection_does_not_treat_a_deleted_bookmark_source_as_selected(
+ monkeypatch,
+):
+ class EmptyResult:
+ def all(self):
+ return []
+
+ class EmptySession:
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ return None
+
+ def exec(self, statement):
+ return EmptyResult()
+
+ monkeypatch.setattr(setup, "SessionLocal", EmptySession)
+
+ source_id, sources, ready = setup._source_and_job_status(
+ connected_state().model_copy(update={"bookmark_source_id": 25})
+ )
+
+ assert source_id is None
+ assert sources == ()
+ assert ready is False
+
+
+def test_setup_routes_replace_the_old_authorize_and_bookmark_shortcuts():
+ import fastapi
+
+ router = fastapi.APIRouter()
+ Extension._register_apis(router)
+ routes = {route.path: route for route in router.routes if isinstance(route, APIRoute)}
+ paths = routes.keys()
+
+ assert {
+ "/setup",
+ "/setup/oauth-app",
+ "/setup/oauth-transactions",
+ "/setup/oauth-transactions/{transaction_id}",
+ "/setup/account",
+ "/setup/bookmark-source",
+ "/setup/finish",
+ "/auth/callback",
+ } <= paths
+ assert "/auth/authorize" not in paths
+ assert "/bookmark" not in paths
+ assert routes["/setup/oauth-transactions"].status_code == 201
+ assert [(item.method, item.path) for item in Extension.public_http_routes()] == [
+ ("GET", "/auth/callback")
+ ]
+
+
+def test_on_close_does_not_write_a_stale_config_snapshot(monkeypatch):
+ config, _ = attach_runtime(TwitterExtensionConfig(api_language="zh-CN"))
async def fail_close():
raise RuntimeError("Twitter cleanup failed")
monkeypatch.setattr(TwitterAPI, "close_singleton", fail_close)
- try:
- with pytest.raises(RuntimeError, match="Twitter cleanup failed"):
- asyncio.run(Extension.on_close())
- finally:
- Extension.release_runtime()
+ with pytest.raises(RuntimeError, match="Twitter cleanup failed"):
+ asyncio.run(Extension.on_close())
+
+ assert config == TwitterExtensionConfig(api_language="zh-CN").model_dump(mode="json")
- assert persisted == [TwitterExtensionConfig(api_language="zh-CN").model_dump()]
+
+def test_cookie_converter_reads_sensitive_input_from_stdin():
+ result = subprocess.run( # noqa: S603
+ [sys.executable, str(COOKIE_CONVERTER)],
+ input="session=secret; preference=compact",
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+
+ assert result.returncode == 0
+ assert json.loads(result.stdout) == {
+ "session": "secret",
+ "preference": "compact",
+ }
def test_twikit_cookie_persistence_creates_runtime_directory(monkeypatch, tmp_path):
diff --git a/tests/migrations/test_extension_peer_enabled_rpc.py b/tests/migrations/test_extension_peer_enabled_rpc.py
index 1ca2869..7b3c454 100644
--- a/tests/migrations/test_extension_peer_enabled_rpc.py
+++ b/tests/migrations/test_extension_peer_enabled_rpc.py
@@ -17,7 +17,7 @@
import sqlmodel
from app.business.extension.errors import ExtensionStateConflictError
-from app.business.extension.state import SQLExtensionStateStore
+from app.business.extension.state import SQLExtensionStore
@pytest.fixture(scope="module")
@@ -69,16 +69,21 @@ def postgres_connection_info(tmp_path_factory) -> Iterator[dict[str, object]]:
@pytest.fixture(scope="module")
def rpc_database(postgres_connection_info, monkeypatch_module):
- migration = importlib.import_module(
+ native_migration = importlib.import_module(
"migrations.versions.b8c1d2e3f4a5_native_extension_distribution_cutover"
)
+ state_migration = importlib.import_module(
+ "migrations.versions.c7d8e9f0a1b2_add_extension_state"
+ )
guard_statements: list[str] = []
- monkeypatch_module.setattr(migration.op, "execute", guard_statements.append)
- migration._create_extension_state_guard()
- assert len(guard_statements) == 2
+ monkeypatch_module.setattr(native_migration.op, "execute", guard_statements.append)
+ native_migration._create_extension_state_guard()
+ monkeypatch_module.setattr(state_migration.op, "execute", guard_statements.append)
+ state_migration._replace_state_guard(include_extension_state=True)
+ assert len(guard_statements) == 3
statements: list[str] = []
- monkeypatch_module.setattr(migration.op, "execute", statements.append)
- migration._create_peer_enable_rpc()
+ monkeypatch_module.setattr(native_migration.op, "execute", statements.append)
+ native_migration._create_peer_enable_rpc()
assert len(statements) == 2
with psycopg.connect(**postgres_connection_info, autocommit=True) as connection:
connection.execute("CREATE SCHEMA inkcre")
@@ -94,6 +99,7 @@ def rpc_database(postgres_connection_info, monkeypatch_module):
enabled uuid[] NOT NULL DEFAULT '{}',
nickname text,
config jsonb NOT NULL DEFAULT '{}',
+ state jsonb NOT NULL DEFAULT '{}',
config_schema jsonb
)
"""
@@ -239,6 +245,16 @@ def test_authenticated_cannot_bypass_rpc_or_insert_enabled_intent(rpc_database):
"UPDATE inkcre.extensions SET config = '{\"safe\": true}'::jsonb "
"WHERE name = 'inkcre/test'"
)
+ with pytest.raises(psycopg.errors.InsufficientPrivilege):
+ connection.execute(
+ "UPDATE inkcre.extensions SET state = '{\"unsafe\": true}'::jsonb "
+ "WHERE name = 'inkcre/test'"
+ )
+ with pytest.raises(psycopg.errors.CheckViolation, match="state must be empty"):
+ connection.execute(
+ "INSERT INTO inkcre.extensions (name, version, state) "
+ "VALUES ('inkcre/state-injected', '1.0.0', '{\"unsafe\": true}'::jsonb)"
+ )
connection.execute("RESET ROLE")
@@ -262,6 +278,24 @@ def test_database_blocks_version_change_and_delete_while_enabled(rpc_database):
)
+def test_database_blocks_version_change_while_extension_state_is_not_empty(
+ rpc_database,
+):
+ with psycopg.connect(**rpc_database, autocommit=True) as connection:
+ connection.execute(
+ "INSERT INTO inkcre.extensions (name, version, state) "
+ "VALUES ('inkcre/stateful', '1.0.0', '{}')"
+ )
+ connection.execute(
+ "UPDATE inkcre.extensions SET state = '{\"authorization\": true}'::jsonb "
+ "WHERE name = 'inkcre/stateful'"
+ )
+ with pytest.raises(psycopg.errors.CheckViolation, match="state is not empty"):
+ connection.execute(
+ "UPDATE inkcre.extensions SET version = '2.0.0' WHERE name = 'inkcre/stateful'"
+ )
+
+
def test_internal_guard_is_not_in_the_postgrest_protocol_schema(rpc_database):
with psycopg.connect(**rpc_database, autocommit=True) as connection:
functions = connection.execute(
@@ -288,7 +322,7 @@ def test_concurrent_first_install_returns_semantic_conflict(rpc_database):
def make_session() -> sqlmodel.Session:
return sqlmodel.Session(engine)
- store = SQLExtensionStateStore(make_session)
+ store = SQLExtensionStore(make_session)
with psycopg.connect(**rpc_database) as first:
first.execute(
"SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
diff --git a/tests/migrations/test_extension_registry_schema.py b/tests/migrations/test_extension_registry_schema.py
index 72371b8..b107e3a 100644
--- a/tests/migrations/test_extension_registry_schema.py
+++ b/tests/migrations/test_extension_registry_schema.py
@@ -19,6 +19,7 @@ def test_canonical_extensions_relation_is_the_only_extension_state_model():
table.c.enabled,
table.c.nickname,
table.c.config,
+ table.c.state,
table.c.config_schema,
]
assert list(table.primary_key.columns) == [table.c.name]
@@ -27,6 +28,7 @@ def test_canonical_extensions_relation_is_the_only_extension_state_model():
assert not table.c.version.nullable
assert not table.c.enabled.nullable
assert not table.c.config.nullable
+ assert not table.c.state.nullable
assert "extension_installations" not in {
candidate.name for candidate in get_target_metadata().tables.values()
}
@@ -46,4 +48,5 @@ def test_canonical_extensions_constraints_match_public_validators():
assert constraints == {
"extensions_name_canonical": f"name ~ '{EXTENSION_NAME_PATTERN}'",
"extensions_version_canonical": f"version ~ '{EXTENSION_SEMVER_PATTERN}'",
+ "extensions_state_object": "jsonb_typeof(state) = 'object'",
}
diff --git a/tests/migrations/test_metadata.py b/tests/migrations/test_metadata.py
index 1eaa40e..ed94ae3 100644
--- a/tests/migrations/test_metadata.py
+++ b/tests/migrations/test_metadata.py
@@ -62,6 +62,7 @@ def test_production_required_columns_are_not_nullable():
("extensions", "config"),
("extensions", "enabled"),
("extensions", "name"),
+ ("extensions", "state"),
("extensions", "version"),
("logs", "timestamp"),
("sources", "config"),
diff --git a/tests/test_deployment_profile.py b/tests/test_deployment_profile.py
index cc08bd8..3f78f20 100644
--- a/tests/test_deployment_profile.py
+++ b/tests/test_deployment_profile.py
@@ -23,7 +23,7 @@ def test_production_profile_projects_the_executable_contract():
assert profile["format"] == 1
assert profile["environment"] == "production"
assert profile["database_contract"] == {
- "migration_head": "b8c1d2e3f4a5",
+ "migration_head": "c7d8e9f0a1b2",
"protocol_schema": PROTOCOL_SCHEMA,
"revision": CONTRACT_REVISION,
}
diff --git a/tests/test_dev_database.py b/tests/test_dev_database.py
index 91c10f8..eb8f791 100644
--- a/tests/test_dev_database.py
+++ b/tests/test_dev_database.py
@@ -134,3 +134,34 @@ def test_source_fingerprint_names_and_hashes_each_file(monkeypatch, tmp_path):
compose.write_text("services:\n postgres: {}")
assert dev_database._source_fingerprint() != initial
+
+
+def test_readiness_invokes_the_container_cli(monkeypatch):
+ state = _state("0123456789abcdef")
+ calls = []
+
+ def compose(_state, arguments, *, timeout):
+ calls.append((arguments, timeout))
+ return '{"status":"ok"}'
+
+ monkeypatch.setattr(dev_database, "_compose", compose)
+
+ assert dev_database._readiness(state)["status"] == "ok"
+ assert calls == [
+ (
+ (
+ "run",
+ "--rm",
+ "--no-deps",
+ "init",
+ "python",
+ "scripts/container.py",
+ "db",
+ "ready",
+ "--profile",
+ "development",
+ "--json",
+ ),
+ 45,
+ )
+ ]
diff --git a/tests/test_dev_database_provider.py b/tests/test_dev_database_provider.py
index dde5922..488e64c 100644
--- a/tests/test_dev_database_provider.py
+++ b/tests/test_dev_database_provider.py
@@ -62,6 +62,7 @@ def test_remote_payload_contains_only_the_runtime_build_surface(tmp_path):
assert "compose.args" in names
assert "context/Dockerfile" in names
assert "context/app" in names
+ assert "context/release/database-contract/README.md" in names
assert "context/tasks" not in names
assert "context/.env" not in names
assert not any("__pycache__" in Path(name).parts for name in names)
@@ -85,6 +86,14 @@ def test_provider_equality_includes_exact_ssh_target_and_binary():
assert not provider.same_database_provider(first, second)
+def test_control_socket_does_not_inherit_the_worktree_path(tmp_path, monkeypatch):
+ monkeypatch.setattr(provider.tempfile, "gettempdir", lambda: str(tmp_path))
+
+ assert provider.control_socket("0123456789abcdef") == (
+ tmp_path / "inkcre-core-py-ssh" / "0123456789abcdef"
+ )
+
+
def test_compose_failure_is_actionable_and_redacts_runtime_secrets(
monkeypatch,
tmp_path,
diff --git a/tests/test_extension_distribution.py b/tests/test_extension_distribution.py
index 6fe60cf..6a12ce1 100644
--- a/tests/test_extension_distribution.py
+++ b/tests/test_extension_distribution.py
@@ -164,7 +164,7 @@ def test_all_first_party_projects_build_pep420_entry_point_wheels(tmp_path: Path
assert "extensions/__init__.py" not in members
assert f"extensions/{extension}/__init__.py" in members
- assert read_project(PROJECT_ROOT / "extensions/twitter").version == "0.1.1"
+ assert read_project(PROJECT_ROOT / "extensions/twitter").version == "0.2.0"
venv = tmp_path / "lifecycle-venv"
subprocess.run( # noqa: S603 -- fixed interpreter and disposable venv
@@ -238,12 +238,21 @@ def test_all_first_party_projects_build_pep420_entry_point_wheels(tmp_path: Path
app = fastapi.FastAPI()
extension.on_start(
app,
- ExtensionRuntimeRecord(name, {{}}, persisted.append, schemas.append),
+ ExtensionRuntimeRecord(
+ extension_id=name,
+ config={{}},
+ read_config=lambda: {{}},
+ persist_config=persisted.append,
+ read_state=lambda: {{}},
+ mutate_state=lambda transform: transform({{}}),
+ mutate_config_and_state=lambda transform: transform({{}}, {{}}),
+ persist_config_schema=schemas.append,
+ ),
)
assert extension.runtime_active()
assert schemas
asyncio.run(extension.on_close())
- assert persisted
+ assert not persisted
extension.unpublish()
assert not extension.runtime_active()
extension.release_runtime()
@@ -573,7 +582,7 @@ def test_extension_publish_changed_source_uses_the_bumped_release_version():
producer = read_project(PROJECT_ROOT / "extensions/twitter")
assert any(path.startswith("extensions/twitter/") for path in changed_paths)
- assert producer.version == "0.1.1"
+ assert producer.version == "0.2.0"
def test_extension_publish_changed_source_same_version_keeps_registry_conflict_fatal():
diff --git a/tests/test_extension_public_http.py b/tests/test_extension_public_http.py
new file mode 100644
index 0000000..5b1564d
--- /dev/null
+++ b/tests/test_extension_public_http.py
@@ -0,0 +1,42 @@
+"""Public Extension callback authority stays exact and publication-scoped."""
+
+import fastapi
+from fastapi.testclient import TestClient
+import pytest
+
+from app.business.extension.runtime import PublicHTTPRoute, PublicHTTPRouteClaim
+from app.middleware import JWTMiddleware
+
+
+def test_exact_claimed_callback_bypasses_jwt_but_other_methods_do_not():
+ app = fastapi.FastAPI()
+
+ @app.get("/twitter/auth/callback")
+ def callback():
+ return {"ok": True}
+
+ claim = PublicHTTPRouteClaim.acquire(
+ "twitter",
+ (PublicHTTPRoute(method="GET", path="/auth/callback"),),
+ tuple(app.routes),
+ )
+ assert claim is not None
+ app.add_middleware(JWTMiddleware)
+ try:
+ with TestClient(app, raise_server_exceptions=False) as client:
+ assert client.get("/twitter/auth/callback?code=secret").status_code == 200
+ assert client.post("/twitter/auth/callback").status_code != 200
+ assert client.get("/twitter/setup").status_code != 200
+ finally:
+ claim.release()
+
+ assert not PublicHTTPRouteClaim.permits("GET", "/twitter/auth/callback")
+
+
+@pytest.mark.parametrize(
+ "path",
+ ["callback", "/callbacks/{provider}", "/callback*", "/callback?mode=x"],
+)
+def test_public_route_declaration_rejects_non_exact_paths(path: str):
+ with pytest.raises(ValueError, match="exact relative path"):
+ PublicHTTPRoute(method="GET", path=path)
diff --git a/tests/test_extension_registry_runtime.py b/tests/test_extension_registry_runtime.py
index 69fde6b..948641e 100644
--- a/tests/test_extension_registry_runtime.py
+++ b/tests/test_extension_registry_runtime.py
@@ -17,7 +17,7 @@
ExtensionRestartRequiredError,
ExtensionStateConflictError,
)
-from app.business.extension.main import ExtensionBase, ExtensionHost
+from app.business.extension.main import ExtensionBase, ExtensionHost, PublicHTTPRoute
from app.business.extension.release import (
EntryPointDescriptor,
ExtensionReleaseDescriptor,
@@ -25,8 +25,8 @@
require_python_association,
simple_project_and_index_urls,
)
-from app.business.extension.state import ExtensionState
-from app.business.extension.runtime import ExtensionRuntimeClaim
+from app.business.extension.state import InstalledExtension
+from app.business.extension.runtime import ExtensionRuntimeClaim, PublicHTTPRouteClaim
from app.business.client import ClientManager
@@ -38,7 +38,16 @@ class FixtureConfig(sqlmodel.SQLModel):
value: int = 1
-class FixtureExtension(ExtensionBase, ext_id="fixture", config_cls=FixtureConfig):
+class FixtureRuntimeState(sqlmodel.SQLModel):
+ counter: int = 0
+
+
+class FixtureExtension(
+ ExtensionBase,
+ ext_id="fixture",
+ config_cls=FixtureConfig,
+ state_cls=FixtureRuntimeState,
+):
fail_close = False
@classmethod
@@ -47,6 +56,14 @@ def _register_apis(cls, router: fastapi.APIRouter) -> None:
def status():
return {"ok": True}
+ @router.get("/callback")
+ def callback():
+ return {"ok": True}
+
+ @classmethod
+ def public_http_routes(cls):
+ return (PublicHTTPRoute(method="GET", path="/callback"),)
+
@classmethod
async def on_close(cls) -> None:
if cls.fail_close:
@@ -79,8 +96,9 @@ def release(*, state: str = "published", version: str = "1.0.0"):
class FakeStore:
- def __init__(self, state: ExtensionState | None = None) -> None:
+ def __init__(self, state: InstalledExtension | None = None) -> None:
self.state = state
+ self.runtime_state: dict[str, typing.Any] = {}
self.set_calls: list[tuple[str, uuid.UUID, bool]] = []
self.fail_set = False
self.version_on_enable: str | None = None
@@ -94,7 +112,7 @@ def get(self, name: str):
def install(self, name: str, version: str, nickname: str):
if self.state is not None and self.state.version != version and self.state.enabled:
raise ExtensionStateConflictError("enabled")
- self.state = ExtensionState(
+ self.state = InstalledExtension(
name=name,
version=version,
nickname=nickname,
@@ -111,6 +129,24 @@ def update_config(self, name: str, config: dict):
self.state = self.state.model_copy(update={"config": config})
return self.state
+ def read_config(self, name: str):
+ assert self.state is not None
+ return dict(self.state.config)
+
+ def read_state(self, name: str):
+ return dict(self.runtime_state)
+
+ def mutate_state(self, name: str, transform):
+ self.runtime_state = transform(dict(self.runtime_state))
+ return dict(self.runtime_state)
+
+ def mutate_config_and_state(self, name: str, transform):
+ assert self.state is not None
+ config, state = transform(dict(self.state.config), dict(self.runtime_state))
+ self.state = self.state.model_copy(update={"config": config})
+ self.runtime_state = state
+ return dict(config), dict(state)
+
def update_config_schema(self, name: str, schema: dict):
assert self.state is not None
self.state = self.state.model_copy(update={"config_schema": schema})
@@ -183,6 +219,8 @@ def clean_runtime(monkeypatch):
FakeModules.fail_unload = False
with ExtensionRuntimeClaim._lock:
ExtensionRuntimeClaim._owners.clear()
+ with PublicHTTPRouteClaim._lock:
+ PublicHTTPRouteClaim._owners.clear()
monkeypatch.setattr(host_module, "DistributionModules", FakeModules)
monkeypatch.setattr(ClientManager, "get_current_client_id", lambda: PEER_ID)
yield
@@ -191,9 +229,11 @@ def clean_runtime(monkeypatch):
FixtureExtension.release_runtime()
with ExtensionRuntimeClaim._lock:
ExtensionRuntimeClaim._owners.clear()
+ with PublicHTTPRouteClaim._lock:
+ PublicHTTPRouteClaim._owners.clear()
-def make_host(state: ExtensionState, descriptor=None):
+def make_host(state: InstalledExtension, descriptor=None):
store = FakeStore(state)
consumer = FakeConsumer()
host = ExtensionHost(
@@ -205,7 +245,7 @@ def make_host(state: ExtensionState, descriptor=None):
def test_enable_starts_before_atomic_enabled_rpc_and_rolls_back_on_rpc_failure():
- state = ExtensionState(name=NAME, version="1.0.0")
+ state = InstalledExtension(name=NAME, version="1.0.0")
host, store, _ = make_host(state)
store.fail_set = True
app = fastapi.FastAPI()
@@ -218,10 +258,48 @@ def test_enable_starts_before_atomic_enabled_rpc_and_rolls_back_on_rpc_failure()
assert store.set_calls == [(NAME, PEER_ID, True)]
assert host.running == {}
assert len(app.routes) == original_routes
+ assert not PublicHTTPRouteClaim.permits("GET", "/fixture/callback")
+
+
+def test_public_http_route_claim_follows_runtime_publication_lifecycle():
+ state = InstalledExtension(name=NAME, version="1.0.0")
+ host, _, _ = make_host(state)
+
+ asyncio.run(host.enable(NAME, app=fastapi.FastAPI()))
+ assert PublicHTTPRouteClaim.permits("GET", "/fixture/callback")
+ assert not PublicHTTPRouteClaim.permits("POST", "/fixture/callback")
+
+ asyncio.run(host.disable(NAME))
+ assert not PublicHTTPRouteClaim.permits("GET", "/fixture/callback")
+
+
+def test_extension_facing_config_and_state_api_reads_fresh_and_persists_immediately():
+ state = InstalledExtension(name=NAME, version="1.0.0")
+ host, store, _ = make_host(state)
+ asyncio.run(host.enable(NAME, app=fastapi.FastAPI()))
+ assert store.state is not None
+ store.state = store.state.model_copy(update={"config": {"value": 7}})
+
+ assert FixtureExtension.get_config().value == 7
+ updated = FixtureExtension.update_config({"value": 9})
+ runtime_state = typing.cast(
+ FixtureRuntimeState,
+ FixtureExtension.mutate_state(
+ lambda current: FixtureRuntimeState(
+ counter=typing.cast(FixtureRuntimeState, current).counter + 1
+ )
+ ),
+ )
+
+ assert updated.value == 9
+ assert store.state.config == {"value": 9}
+ assert runtime_state.counter == 1
+ assert store.runtime_state == {"counter": 1}
+ asyncio.run(host.disable(NAME))
def test_existing_exact_yanked_release_can_cold_restore_without_mutating_intent():
- state = ExtensionState(name=NAME, version="1.0.0", enabled=(PEER_ID,))
+ state = InstalledExtension(name=NAME, version="1.0.0", enabled=(PEER_ID,))
host, store, consumer = make_host(state, release(state="yanked"))
asyncio.run(host.start_enabled(fastapi.FastAPI()))
@@ -232,7 +310,7 @@ def test_existing_exact_yanked_release_can_cold_restore_without_mutating_intent(
def test_concurrent_version_change_compensates_enabled_and_started_runtime():
- state = ExtensionState(name=NAME, version="1.0.0")
+ state = InstalledExtension(name=NAME, version="1.0.0")
host, store, _ = make_host(state)
store.version_on_enable = "2.0.0"
@@ -252,8 +330,8 @@ def test_concurrent_version_change_compensates_enabled_and_started_runtime():
def test_runtime_claim_blocks_cross_namespace_package_collision_until_release():
first_name = "inkcre/fixture"
second_name = "other/fixture"
- first_state = ExtensionState(name=first_name, version="1.0.0")
- second_state = ExtensionState(name=second_name, version="1.0.0")
+ first_state = InstalledExtension(name=first_name, version="1.0.0")
+ second_state = InstalledExtension(name=second_name, version="1.0.0")
first_descriptor = release().model_copy(update={"name": first_name})
second_descriptor = release().model_copy(update={"name": second_name})
first, _, _ = make_host(first_state, first_descriptor)
@@ -272,7 +350,7 @@ def test_runtime_claim_blocks_cross_namespace_package_collision_until_release():
def test_failed_startup_compensation_releases_runtime_claim(monkeypatch):
- state = ExtensionState(name=NAME, version="1.0.0")
+ state = InstalledExtension(name=NAME, version="1.0.0")
failed, _, _ = make_host(state)
def fail_load(self, extension_base):
@@ -285,13 +363,13 @@ def fail_load(self, extension_base):
def test_new_install_rejects_yanked_and_version_change_rejects_enabled_intent():
- missing = ExtensionState(name=NAME, version="1.0.0", enabled=(PEER_ID,))
+ missing = InstalledExtension(name=NAME, version="1.0.0", enabled=(PEER_ID,))
host, _, _ = make_host(missing, release(version="2.0.0"))
with pytest.raises(ExtensionStateConflictError):
host.install(NAME, "2.0.0")
fresh_host, _, _ = make_host(
- ExtensionState(name="inkcre/another", version="1.0.0"),
+ InstalledExtension(name="inkcre/another", version="1.0.0"),
release(state="yanked"),
)
with pytest.raises(ExtensionCompatibilityError):
@@ -299,7 +377,7 @@ def test_new_install_rejects_yanked_and_version_change_rejects_enabled_intent():
def test_loaded_version_change_requires_restart_even_after_disable():
- state = ExtensionState(name=NAME, version="1.0.0")
+ state = InstalledExtension(name=NAME, version="1.0.0")
host, _, _ = make_host(state, release(version="2.0.0"))
host._loaded_versions[NAME] = "1.0.0"
@@ -308,7 +386,7 @@ def test_loaded_version_change_requires_restart_even_after_disable():
def test_disable_rpc_failure_restarts_exact_prior_runtime_and_preserves_intent():
- state = ExtensionState(name=NAME, version="1.0.0", enabled=(PEER_ID,))
+ state = InstalledExtension(name=NAME, version="1.0.0", enabled=(PEER_ID,))
host, store, _ = make_host(state)
app = fastapi.FastAPI()
asyncio.run(host.start_enabled(app))
@@ -323,7 +401,7 @@ def test_disable_rpc_failure_restarts_exact_prior_runtime_and_preserves_intent()
def test_disable_reports_both_rpc_and_restart_compensation_failures(monkeypatch):
- state = ExtensionState(name=NAME, version="1.0.0", enabled=(PEER_ID,))
+ state = InstalledExtension(name=NAME, version="1.0.0", enabled=(PEER_ID,))
host, store, _ = make_host(state)
asyncio.run(host.start_enabled(fastapi.FastAPI()))
store.fail_set = True
@@ -344,7 +422,7 @@ async def fail_restart(*args, **kwargs):
@pytest.mark.parametrize("phase", ["close", "unload"])
def test_partial_stop_failure_remains_tracked_and_never_mutates_enabled(phase: str):
- state = ExtensionState(name=NAME, version="1.0.0", enabled=(PEER_ID,))
+ state = InstalledExtension(name=NAME, version="1.0.0", enabled=(PEER_ID,))
host, store, _ = make_host(state)
app = fastapi.FastAPI()
asyncio.run(host.start_enabled(app))
@@ -374,7 +452,7 @@ def test_host_sdk_uses_npm_semver_ranges_and_prerelease_rules():
)
with pytest.raises(ExtensionCompatibilityError):
require_python_association(
- descriptor.model_copy(update={"python": association(">=0.1.1-beta.1")})
+ descriptor.model_copy(update={"python": association(">=0.1.2-beta.1")})
)
diff --git a/tests/test_extension_source_lifecycle.py b/tests/test_extension_source_lifecycle.py
index 8c70fb6..9b31436 100644
--- a/tests/test_extension_source_lifecycle.py
+++ b/tests/test_extension_source_lifecycle.py
@@ -238,6 +238,17 @@ async def run_job(cls, job_id):
"_source_rows",
classmethod(lambda cls, selected=None: (row,)),
)
+ monkeypatch.setattr(
+ SourceManager,
+ "_get_source_ins",
+ classmethod(
+ lambda cls, source_id, source_type=None: type(
+ "ScheduledSource",
+ (),
+ {"scheduled_collect_config": lambda self: {}},
+ )()
+ ),
+ )
monkeypatch.setattr(SourceCollectJobManager, "run", classmethod(run_job))
activation = SourceManager.set_up_collect_jobs({row.type})
@@ -248,6 +259,54 @@ async def run_job(cls, job_id):
SourceManager.withdraw_runtime_activation(activation)
+def test_collect_job_ensure_reuses_exact_non_failed_input_under_a_database_lock(
+ monkeypatch,
+):
+ jobs: list[SourceCollectJobModel] = []
+ lock_identities: list[str] = []
+
+ class Result:
+ def all(self):
+ return list(jobs)
+
+ class EnsureSession:
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ return None
+
+ def connection(self):
+ return self
+
+ def execute(self, statement, parameters):
+ lock_identities.append(parameters["identity"])
+
+ def exec(self, statement):
+ return Result()
+
+ def add(self, job):
+ jobs.append(job)
+
+ def commit(self):
+ return None
+
+ def refresh(self, job):
+ job.id = 81
+
+ monkeypatch.setattr(collect_job_module, "SessionLocal", EnsureSession)
+ config = {"authorization_id": "account-1", "full": False, "result_limit": 40}
+
+ first, first_created = SourceCollectJobManager.ensure(41, config)
+ second, second_created = SourceCollectJobManager.ensure(41, config)
+
+ assert first_created is True
+ assert second_created is False
+ assert second is first
+ assert len(jobs) == 1
+ assert len(set(lock_identities)) == 1
+
+
def test_partial_scheduler_activation_rolls_back_added_jobs_and_type_ownership(
monkeypatch,
):