mirror of
https://github.com/open-webui/open-webui.git
synced 2026-07-24 07:14:16 +00:00
Both session factories run with expire_on_commit=False, so ORM objects keep their attribute values after commit. Every session.refresh issued right after a commit therefore re-SELECTed a row whose values the session already held, including full chat JSON blobs and user settings, purely to overwrite identical data. Fifty such calls existed across the model layer, covering nearly every write path in the app (chat inserts, title updates, pin/archive toggles, user role and settings updates, tool, prompt, function, model, file, tag, feedback, memory, automation and grant writes). All fifty are removed. The only refreshes with an actual job were the two update-then-reload paths in tools and skills, where a Core UPDATE statement bypasses the identity map; those now use session.get(..., populate_existing=True), which guarantees a fresh row in one SELECT whether or not the row was already present in the session (the previous code issued get plus refresh, two SELECTs, on the default configuration). Benchmark (real SQLite DB, per write): | write path | before | after | | --- | --- | --- | | chat title update, ~600 KB chat blob | 2.08 ms | 1.24 ms | | user role update, small row | 1.21 ms | 0.68 ms | On Postgres each removed refresh is additionally a network round trip. The chat-blob case also skips re-parsing the entire JSON document per write. Functionally verified against a fresh database: user insert, role and settings updates, chat insert (including the server-default meta column, which is always provided client-side), title update and pin toggle, tool insert and the Core-update reload path, tag insert and the prompt insert flow that pins version_id after history creation all return correct values and persist correctly.
1009 lines
36 KiB
Python
1009 lines
36 KiB
Python
import json
|
|
import secrets
|
|
import time
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from open_webui.internal.db import Base, JSONField, get_async_db_context
|
|
from open_webui.models.access_grants import (
|
|
AccessGrantModel,
|
|
AccessGrants,
|
|
)
|
|
from open_webui.models.groups import Groups
|
|
from open_webui.utils.validate import validate_profile_image_url
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
from sqlalchemy import (
|
|
JSON,
|
|
BigInteger,
|
|
Boolean,
|
|
Column,
|
|
ForeignKey,
|
|
String,
|
|
Text,
|
|
UniqueConstraint,
|
|
and_,
|
|
case,
|
|
delete,
|
|
func,
|
|
or_,
|
|
select,
|
|
update,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
####################
|
|
# Channel DB Schema
|
|
####################
|
|
|
|
|
|
class Channel(Base):
|
|
__tablename__ = 'channel'
|
|
|
|
id = Column(Text, primary_key=True, unique=True)
|
|
user_id = Column(Text)
|
|
type = Column(Text, nullable=True)
|
|
|
|
name = Column(Text)
|
|
description = Column(Text, nullable=True)
|
|
|
|
# Used to indicate if the channel is private (for 'group' type channels)
|
|
is_private = Column(Boolean, nullable=True)
|
|
|
|
data = Column(JSON, nullable=True)
|
|
meta = Column(JSON, nullable=True)
|
|
|
|
created_at = Column(BigInteger)
|
|
|
|
updated_at = Column(BigInteger)
|
|
updated_by = Column(Text, nullable=True)
|
|
|
|
archived_at = Column(BigInteger, nullable=True)
|
|
archived_by = Column(Text, nullable=True)
|
|
|
|
deleted_at = Column(BigInteger, nullable=True)
|
|
deleted_by = Column(Text, nullable=True)
|
|
|
|
|
|
class ChannelModel(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: str
|
|
user_id: str
|
|
|
|
type: Optional[str] = None
|
|
|
|
name: str
|
|
description: Optional[str] = None
|
|
|
|
is_private: Optional[bool] = None
|
|
|
|
data: Optional[dict] = None
|
|
meta: Optional[dict] = None
|
|
access_grants: list[AccessGrantModel] = Field(default_factory=list)
|
|
|
|
created_at: int # timestamp in epoch (time_ns)
|
|
|
|
updated_at: int # timestamp in epoch (time_ns)
|
|
updated_by: Optional[str] = None
|
|
|
|
archived_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
archived_by: Optional[str] = None
|
|
|
|
deleted_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
deleted_by: Optional[str] = None
|
|
|
|
|
|
class ChannelMember(Base):
|
|
__tablename__ = 'channel_member'
|
|
|
|
id = Column(Text, primary_key=True, unique=True)
|
|
channel_id = Column(Text, nullable=False)
|
|
user_id = Column(Text, nullable=False)
|
|
|
|
role = Column(Text, nullable=True)
|
|
status = Column(Text, nullable=True)
|
|
|
|
is_active = Column(Boolean, nullable=False, default=True)
|
|
|
|
is_channel_muted = Column(Boolean, nullable=False, default=False)
|
|
is_channel_pinned = Column(Boolean, nullable=False, default=False)
|
|
|
|
data = Column(JSON, nullable=True)
|
|
meta = Column(JSON, nullable=True)
|
|
|
|
invited_at = Column(BigInteger, nullable=True)
|
|
invited_by = Column(Text, nullable=True)
|
|
|
|
joined_at = Column(BigInteger)
|
|
left_at = Column(BigInteger, nullable=True)
|
|
|
|
last_read_at = Column(BigInteger, nullable=True)
|
|
|
|
created_at = Column(BigInteger)
|
|
updated_at = Column(BigInteger)
|
|
|
|
|
|
class ChannelMemberModel(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: str
|
|
channel_id: str
|
|
user_id: str
|
|
|
|
role: Optional[str] = None
|
|
status: Optional[str] = None
|
|
|
|
is_active: bool = True
|
|
|
|
is_channel_muted: bool = False
|
|
is_channel_pinned: bool = False
|
|
|
|
data: Optional[dict] = None
|
|
meta: Optional[dict] = None
|
|
|
|
invited_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
invited_by: Optional[str] = None
|
|
|
|
joined_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
left_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
|
|
last_read_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
|
|
created_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
updated_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
|
|
|
|
class ChannelFile(Base):
|
|
__tablename__ = 'channel_file'
|
|
|
|
id = Column(Text, unique=True, primary_key=True)
|
|
user_id = Column(Text, nullable=False)
|
|
|
|
channel_id = Column(Text, ForeignKey('channel.id', ondelete='CASCADE'), nullable=False)
|
|
message_id = Column(Text, ForeignKey('message.id', ondelete='CASCADE'), nullable=True)
|
|
file_id = Column(Text, ForeignKey('file.id', ondelete='CASCADE'), nullable=False)
|
|
|
|
created_at = Column(BigInteger, nullable=False)
|
|
updated_at = Column(BigInteger, nullable=False)
|
|
|
|
__table_args__ = (UniqueConstraint('channel_id', 'file_id', name='uq_channel_file_channel_file'),)
|
|
|
|
|
|
class ChannelFileModel(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: str
|
|
|
|
channel_id: str
|
|
file_id: str
|
|
user_id: str
|
|
|
|
created_at: int # timestamp in epoch (time_ns)
|
|
updated_at: int # timestamp in epoch (time_ns)
|
|
|
|
|
|
class ChannelWebhook(Base):
|
|
__tablename__ = 'channel_webhook'
|
|
|
|
id = Column(Text, primary_key=True, unique=True)
|
|
channel_id = Column(Text, nullable=False)
|
|
user_id = Column(Text, nullable=False)
|
|
|
|
name = Column(Text, nullable=False)
|
|
profile_image_url = Column(Text, nullable=True)
|
|
|
|
token = Column(Text, nullable=False)
|
|
last_used_at = Column(BigInteger, nullable=True)
|
|
|
|
created_at = Column(BigInteger, nullable=False)
|
|
updated_at = Column(BigInteger, nullable=False)
|
|
|
|
|
|
class ChannelWebhookModel(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: str
|
|
channel_id: str
|
|
user_id: str
|
|
|
|
name: str
|
|
profile_image_url: Optional[str] = None
|
|
|
|
token: str
|
|
last_used_at: Optional[int] = None # timestamp in epoch (time_ns)
|
|
|
|
created_at: int # timestamp in epoch (time_ns)
|
|
updated_at: int # timestamp in epoch (time_ns)
|
|
|
|
|
|
####################
|
|
# Forms
|
|
####################
|
|
|
|
|
|
class ChannelResponse(ChannelModel):
|
|
is_manager: bool = False
|
|
write_access: bool = False
|
|
|
|
user_count: Optional[int] = None
|
|
|
|
|
|
class ChannelForm(BaseModel):
|
|
name: str = ''
|
|
description: Optional[str] = None
|
|
is_private: Optional[bool] = None
|
|
data: Optional[dict] = None
|
|
meta: Optional[dict] = None
|
|
access_grants: Optional[list[dict]] = None
|
|
group_ids: Optional[list[str]] = None
|
|
user_ids: Optional[list[str]] = None
|
|
|
|
|
|
class CreateChannelForm(ChannelForm):
|
|
type: Optional[str] = None
|
|
|
|
|
|
class ChannelWebhookForm(BaseModel):
|
|
name: str
|
|
profile_image_url: Optional[str] = None
|
|
|
|
@field_validator('profile_image_url', mode='before')
|
|
@classmethod
|
|
def check_profile_image_url(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return v
|
|
return validate_profile_image_url(v)
|
|
|
|
|
|
class ChannelTable:
|
|
async def _get_access_grants(self, channel_id: str, db: Optional[AsyncSession] = None) -> list[AccessGrantModel]:
|
|
return await AccessGrants.get_grants_by_resource('channel', channel_id, db=db)
|
|
|
|
async def _to_channel_model(
|
|
self,
|
|
channel: Channel,
|
|
access_grants: Optional[list[AccessGrantModel]] = None,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> ChannelModel:
|
|
channel_data = ChannelModel.model_validate(channel).model_dump(exclude={'access_grants'})
|
|
channel_data['access_grants'] = (
|
|
access_grants if access_grants is not None else await self._get_access_grants(channel_data['id'], db=db)
|
|
)
|
|
return ChannelModel.model_validate(channel_data)
|
|
|
|
async def _collect_unique_user_ids(
|
|
self,
|
|
invited_by: str,
|
|
user_ids: Optional[list[str]] = None,
|
|
group_ids: Optional[list[str]] = None,
|
|
) -> set[str]:
|
|
"""
|
|
Collect unique user ids from:
|
|
- invited_by
|
|
- user_ids
|
|
- each group in group_ids
|
|
Returns a set for efficient SQL diffing.
|
|
"""
|
|
users = set(user_ids or [])
|
|
users.add(invited_by)
|
|
|
|
for group_id in group_ids or []:
|
|
group_user_ids = await Groups.get_group_user_ids_by_id(group_id)
|
|
users.update(group_user_ids)
|
|
|
|
return users
|
|
|
|
def _create_membership_models(
|
|
self,
|
|
channel_id: str,
|
|
invited_by: str,
|
|
user_ids: set[str],
|
|
) -> list[ChannelMember]:
|
|
"""
|
|
Takes a set of NEW user IDs (already filtered to exclude existing members).
|
|
Returns ORM ChannelMember objects to be added.
|
|
"""
|
|
now = int(time.time_ns())
|
|
memberships = []
|
|
|
|
for uid in user_ids:
|
|
model = ChannelMemberModel(
|
|
**{
|
|
'id': str(uuid.uuid4()),
|
|
'channel_id': channel_id,
|
|
'user_id': uid,
|
|
'status': 'joined',
|
|
'is_active': True,
|
|
'is_channel_muted': False,
|
|
'is_channel_pinned': False,
|
|
'invited_at': now,
|
|
'invited_by': invited_by,
|
|
'joined_at': now,
|
|
'left_at': None,
|
|
'last_read_at': now,
|
|
'created_at': now,
|
|
'updated_at': now,
|
|
}
|
|
)
|
|
memberships.append(ChannelMember(**model.model_dump()))
|
|
|
|
return memberships
|
|
|
|
def _has_permission(self, db, query, filter: dict, permission: str = 'read'):
|
|
return AccessGrants.has_permission_filter(
|
|
db=db,
|
|
query=query,
|
|
DocumentModel=Channel,
|
|
filter=filter,
|
|
resource_type='channel',
|
|
permission=permission,
|
|
)
|
|
|
|
async def insert_new_channel(
|
|
self, form_data: CreateChannelForm, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
channel = ChannelModel(
|
|
**{
|
|
**form_data.model_dump(exclude={'access_grants'}),
|
|
'type': form_data.type if form_data.type else None,
|
|
'name': form_data.name.lower(),
|
|
'id': str(uuid.uuid4()),
|
|
'user_id': user_id,
|
|
'created_at': int(time.time_ns()),
|
|
'updated_at': int(time.time_ns()),
|
|
'access_grants': [],
|
|
}
|
|
)
|
|
new_channel = Channel(**channel.model_dump(exclude={'access_grants'}))
|
|
|
|
if form_data.type in ['group', 'dm']:
|
|
users = await self._collect_unique_user_ids(
|
|
invited_by=user_id,
|
|
user_ids=form_data.user_ids,
|
|
group_ids=form_data.group_ids,
|
|
)
|
|
memberships = self._create_membership_models(
|
|
channel_id=new_channel.id,
|
|
invited_by=user_id,
|
|
user_ids=users,
|
|
)
|
|
|
|
db.add_all(memberships)
|
|
db.add(new_channel)
|
|
await db.commit()
|
|
await AccessGrants.set_access_grants('channel', new_channel.id, form_data.access_grants, db=db)
|
|
return await self._to_channel_model(new_channel, db=db)
|
|
|
|
async def get_channels(self, db: Optional[AsyncSession] = None) -> list[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(Channel))
|
|
channels = result.scalars().all()
|
|
channel_ids = [channel.id for channel in channels]
|
|
grants_map = await AccessGrants.get_grants_by_resources('channel', channel_ids, db=db)
|
|
return [
|
|
await self._to_channel_model(
|
|
channel,
|
|
access_grants=grants_map.get(channel.id, []),
|
|
db=db,
|
|
)
|
|
for channel in channels
|
|
]
|
|
|
|
async def get_channels_by_user_id(self, user_id: str, db: Optional[AsyncSession] = None) -> list[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
user_group_ids = [group.id for group in await Groups.get_groups_by_member_id(user_id, db=db)]
|
|
|
|
result = await db.execute(
|
|
select(Channel)
|
|
.join(ChannelMember, Channel.id == ChannelMember.channel_id)
|
|
.filter(
|
|
Channel.deleted_at.is_(None),
|
|
Channel.archived_at.is_(None),
|
|
Channel.type.in_(['group', 'dm']),
|
|
ChannelMember.user_id == user_id,
|
|
ChannelMember.is_active.is_(True),
|
|
)
|
|
)
|
|
membership_channels = result.scalars().all()
|
|
|
|
stmt = select(Channel).filter(
|
|
Channel.deleted_at.is_(None),
|
|
Channel.archived_at.is_(None),
|
|
or_(
|
|
Channel.type.is_(None), # True NULL/None
|
|
Channel.type == '', # Empty string
|
|
and_(Channel.type != 'group', Channel.type != 'dm'),
|
|
),
|
|
)
|
|
stmt = self._has_permission(db, stmt, {'user_id': user_id, 'group_ids': user_group_ids})
|
|
|
|
result = await db.execute(stmt)
|
|
standard_channels = result.scalars().all()
|
|
|
|
all_channels = list(membership_channels) + list(standard_channels)
|
|
channel_ids = [c.id for c in all_channels]
|
|
grants_map = await AccessGrants.get_grants_by_resources('channel', channel_ids, db=db)
|
|
return [
|
|
await self._to_channel_model(c, access_grants=grants_map.get(c.id, []), db=db) for c in all_channels
|
|
]
|
|
|
|
async def get_dm_channel_by_user_ids(
|
|
self, user_ids: list[str], db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
# Ensure uniqueness in case a list with duplicates is passed
|
|
unique_user_ids = list(set(user_ids))
|
|
|
|
match_count = func.sum(
|
|
case(
|
|
(ChannelMember.user_id.in_(unique_user_ids), 1),
|
|
else_=0,
|
|
)
|
|
)
|
|
|
|
subquery = (
|
|
select(ChannelMember.channel_id)
|
|
.group_by(ChannelMember.channel_id)
|
|
# 1. Channel must have exactly len(user_ids) members
|
|
.having(func.count(ChannelMember.user_id) == len(unique_user_ids))
|
|
# 2. All those members must be in unique_user_ids
|
|
.having(match_count == len(unique_user_ids))
|
|
.subquery()
|
|
)
|
|
|
|
result = await db.execute(
|
|
select(Channel)
|
|
.filter(
|
|
Channel.id.in_(select(subquery.c.channel_id)),
|
|
Channel.type == 'dm',
|
|
)
|
|
.limit(1)
|
|
)
|
|
channel = result.scalars().first()
|
|
|
|
return await self._to_channel_model(channel, db=db) if channel else None
|
|
|
|
async def add_members_to_channel(
|
|
self,
|
|
channel_id: str,
|
|
invited_by: str,
|
|
user_ids: Optional[list[str]] = None,
|
|
group_ids: Optional[list[str]] = None,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> list[ChannelMemberModel]:
|
|
async with get_async_db_context(db) as db:
|
|
# 1. Collect all user_ids including groups + inviter
|
|
requested_users = await self._collect_unique_user_ids(invited_by, user_ids, group_ids)
|
|
|
|
result = await db.execute(select(ChannelMember.user_id).filter(ChannelMember.channel_id == channel_id))
|
|
existing_users = {row[0] for row in result.all()}
|
|
|
|
new_user_ids = requested_users - existing_users
|
|
if not new_user_ids:
|
|
return [] # Nothing to add
|
|
|
|
new_memberships = self._create_membership_models(channel_id, invited_by, new_user_ids)
|
|
|
|
db.add_all(new_memberships)
|
|
await db.commit()
|
|
|
|
return [ChannelMemberModel.model_validate(membership) for membership in new_memberships]
|
|
|
|
async def remove_members_from_channel(
|
|
self,
|
|
channel_id: str,
|
|
user_ids: list[str],
|
|
db: Optional[AsyncSession] = None,
|
|
) -> int:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
delete(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id.in_(user_ids),
|
|
)
|
|
)
|
|
await db.commit()
|
|
return result.rowcount # number of rows deleted
|
|
|
|
async def is_user_channel_manager(self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(Channel).filter(Channel.id == channel_id))
|
|
channel = result.scalars().first()
|
|
if channel and channel.user_id == user_id:
|
|
return True
|
|
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
ChannelMember.is_active.is_(True),
|
|
ChannelMember.role == 'manager',
|
|
)
|
|
)
|
|
membership = result.scalars().first()
|
|
return membership is not None
|
|
|
|
async def join_channel(
|
|
self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelMemberModel]:
|
|
async with get_async_db_context(db) as db:
|
|
# Check if the membership already exists
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
)
|
|
)
|
|
existing_membership = result.scalars().first()
|
|
if existing_membership:
|
|
return ChannelMemberModel.model_validate(existing_membership)
|
|
|
|
# Create new membership
|
|
channel_member = ChannelMemberModel(
|
|
**{
|
|
'id': str(uuid.uuid4()),
|
|
'channel_id': channel_id,
|
|
'user_id': user_id,
|
|
'status': 'joined',
|
|
'is_active': True,
|
|
'is_channel_muted': False,
|
|
'is_channel_pinned': False,
|
|
'joined_at': int(time.time_ns()),
|
|
'left_at': None,
|
|
'last_read_at': int(time.time_ns()),
|
|
'created_at': int(time.time_ns()),
|
|
'updated_at': int(time.time_ns()),
|
|
}
|
|
)
|
|
new_membership = ChannelMember(**channel_member.model_dump())
|
|
|
|
db.add(new_membership)
|
|
await db.commit()
|
|
return channel_member
|
|
|
|
async def leave_channel(self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
)
|
|
)
|
|
membership = result.scalars().first()
|
|
if not membership:
|
|
return False
|
|
|
|
membership.status = 'left'
|
|
membership.is_active = False
|
|
membership.left_at = int(time.time_ns())
|
|
membership.updated_at = int(time.time_ns())
|
|
|
|
await db.commit()
|
|
return True
|
|
|
|
async def get_member_by_channel_and_user_id(
|
|
self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelMemberModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
)
|
|
)
|
|
membership = result.scalars().first()
|
|
return ChannelMemberModel.model_validate(membership) if membership else None
|
|
|
|
async def get_members_by_channel_id(
|
|
self, channel_id: str, db: Optional[AsyncSession] = None
|
|
) -> list[ChannelMemberModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelMember).filter(ChannelMember.channel_id == channel_id))
|
|
memberships = result.scalars().all()
|
|
return [ChannelMemberModel.model_validate(membership) for membership in memberships]
|
|
|
|
async def pin_channel(
|
|
self,
|
|
channel_id: str,
|
|
user_id: str,
|
|
is_pinned: bool,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
)
|
|
)
|
|
membership = result.scalars().first()
|
|
if not membership:
|
|
return False
|
|
|
|
membership.is_channel_pinned = is_pinned
|
|
membership.updated_at = int(time.time_ns())
|
|
|
|
await db.commit()
|
|
return True
|
|
|
|
async def update_member_last_read_at(
|
|
self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
)
|
|
)
|
|
membership = result.scalars().first()
|
|
if not membership:
|
|
return False
|
|
|
|
membership.last_read_at = int(time.time_ns())
|
|
membership.updated_at = int(time.time_ns())
|
|
|
|
await db.commit()
|
|
return True
|
|
|
|
async def update_member_active_status(
|
|
self,
|
|
channel_id: str,
|
|
user_id: str,
|
|
is_active: bool,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelMember).filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
)
|
|
)
|
|
membership = result.scalars().first()
|
|
if not membership:
|
|
return False
|
|
|
|
membership.is_active = is_active
|
|
membership.updated_at = int(time.time_ns())
|
|
|
|
await db.commit()
|
|
return True
|
|
|
|
async def is_user_channel_member(self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelMember)
|
|
.filter(
|
|
ChannelMember.channel_id == channel_id,
|
|
ChannelMember.user_id == user_id,
|
|
ChannelMember.is_active.is_(True),
|
|
)
|
|
.limit(1)
|
|
)
|
|
membership = result.scalars().first()
|
|
return membership is not None
|
|
|
|
async def get_channel_by_id(self, id: str, db: Optional[AsyncSession] = None) -> Optional[ChannelModel]:
|
|
try:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(Channel).filter(Channel.id == id))
|
|
channel = result.scalars().first()
|
|
return await self._to_channel_model(channel, db=db) if channel else None
|
|
except Exception:
|
|
return None
|
|
|
|
async def get_channels_by_file_id(self, file_id: str, db: Optional[AsyncSession] = None) -> list[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelFile).filter(ChannelFile.file_id == file_id))
|
|
channel_files = result.scalars().all()
|
|
channel_ids = [cf.channel_id for cf in channel_files]
|
|
result = await db.execute(select(Channel).filter(Channel.id.in_(channel_ids)))
|
|
channels = result.scalars().all()
|
|
grants_map = await AccessGrants.get_grants_by_resources('channel', channel_ids, db=db)
|
|
return [
|
|
await self._to_channel_model(
|
|
channel,
|
|
access_grants=grants_map.get(channel.id, []),
|
|
db=db,
|
|
)
|
|
for channel in channels
|
|
]
|
|
|
|
async def get_channels_by_file_id_and_user_id(
|
|
self, file_id: str, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> list[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
# 1. Determine which channels have this file
|
|
result = await db.execute(select(ChannelFile).filter(ChannelFile.file_id == file_id))
|
|
channel_file_rows = result.scalars().all()
|
|
channel_ids = [row.channel_id for row in channel_file_rows]
|
|
|
|
if not channel_ids:
|
|
return []
|
|
|
|
# 2. Load all channel rows that still exist
|
|
result = await db.execute(
|
|
select(Channel).filter(
|
|
Channel.id.in_(channel_ids),
|
|
Channel.deleted_at.is_(None),
|
|
Channel.archived_at.is_(None),
|
|
)
|
|
)
|
|
channels = result.scalars().all()
|
|
if not channels:
|
|
return []
|
|
|
|
# Preload user's group membership
|
|
user_group_ids = [g.id for g in await Groups.get_groups_by_member_id(user_id, db=db)]
|
|
|
|
allowed_channels = []
|
|
|
|
for channel in channels:
|
|
# --- Case A: group or dm => user must be an active member ---
|
|
if channel.type in ['group', 'dm']:
|
|
result = await db.execute(
|
|
select(ChannelMember)
|
|
.filter(
|
|
ChannelMember.channel_id == channel.id,
|
|
ChannelMember.user_id == user_id,
|
|
ChannelMember.is_active.is_(True),
|
|
)
|
|
.limit(1)
|
|
)
|
|
membership = result.scalars().first()
|
|
if membership:
|
|
allowed_channels.append(await self._to_channel_model(channel, db=db))
|
|
continue
|
|
|
|
# --- Case B: standard channel => rely on ACL permissions ---
|
|
stmt = select(Channel).filter(Channel.id == channel.id)
|
|
|
|
stmt = self._has_permission(
|
|
db,
|
|
stmt,
|
|
{'user_id': user_id, 'group_ids': user_group_ids},
|
|
permission='read',
|
|
)
|
|
|
|
result = await db.execute(stmt)
|
|
allowed = result.scalars().first()
|
|
if allowed:
|
|
allowed_channels.append(await self._to_channel_model(allowed, db=db))
|
|
|
|
return allowed_channels
|
|
|
|
async def get_channel_by_id_and_user_id(
|
|
self, id: str, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
# Fetch the channel
|
|
result = await db.execute(
|
|
select(Channel).filter(
|
|
Channel.id == id,
|
|
Channel.deleted_at.is_(None),
|
|
Channel.archived_at.is_(None),
|
|
)
|
|
)
|
|
channel = result.scalars().first()
|
|
|
|
if not channel:
|
|
return None
|
|
|
|
# If the channel is a group or dm, read access requires membership (active)
|
|
if channel.type in ['group', 'dm']:
|
|
result = await db.execute(
|
|
select(ChannelMember)
|
|
.filter(
|
|
ChannelMember.channel_id == id,
|
|
ChannelMember.user_id == user_id,
|
|
ChannelMember.is_active.is_(True),
|
|
)
|
|
.limit(1)
|
|
)
|
|
membership = result.scalars().first()
|
|
if membership:
|
|
return await self._to_channel_model(channel, db=db)
|
|
else:
|
|
return None
|
|
|
|
# For channels that are NOT group/dm, fall back to ACL-based read access
|
|
stmt = select(Channel).filter(Channel.id == id)
|
|
|
|
# Determine user groups
|
|
user_group_ids = [group.id for group in await Groups.get_groups_by_member_id(user_id, db=db)]
|
|
|
|
# Apply ACL rules
|
|
stmt = self._has_permission(
|
|
db,
|
|
stmt,
|
|
{'user_id': user_id, 'group_ids': user_group_ids},
|
|
permission='read',
|
|
)
|
|
|
|
result = await db.execute(stmt)
|
|
channel_allowed = result.scalars().first()
|
|
return await self._to_channel_model(channel_allowed, db=db) if channel_allowed else None
|
|
|
|
async def update_channel_by_id(
|
|
self, id: str, form_data: ChannelForm, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(Channel).filter(Channel.id == id))
|
|
channel = result.scalars().first()
|
|
if not channel:
|
|
return None
|
|
|
|
channel.name = form_data.name
|
|
channel.description = form_data.description
|
|
channel.is_private = form_data.is_private
|
|
|
|
channel.data = form_data.data
|
|
channel.meta = form_data.meta
|
|
|
|
if form_data.access_grants is not None:
|
|
await AccessGrants.set_access_grants('channel', id, form_data.access_grants, db=db)
|
|
channel.updated_at = int(time.time_ns())
|
|
|
|
await db.commit()
|
|
return await self._to_channel_model(channel, db=db) if channel else None
|
|
|
|
async def add_file_to_channel_by_id(
|
|
self, channel_id: str, file_id: str, user_id: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelFileModel]:
|
|
async with get_async_db_context(db) as db:
|
|
channel_file = ChannelFileModel(
|
|
**{
|
|
'id': str(uuid.uuid4()),
|
|
'channel_id': channel_id,
|
|
'file_id': file_id,
|
|
'user_id': user_id,
|
|
'created_at': int(time.time()),
|
|
'updated_at': int(time.time()),
|
|
}
|
|
)
|
|
|
|
try:
|
|
result = ChannelFile(**channel_file.model_dump())
|
|
db.add(result)
|
|
await db.commit()
|
|
if result:
|
|
return ChannelFileModel.model_validate(result)
|
|
else:
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
async def set_file_message_id_in_channel_by_id(
|
|
self,
|
|
channel_id: str,
|
|
file_id: str,
|
|
message_id: str,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> bool:
|
|
try:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelFile).filter_by(channel_id=channel_id, file_id=file_id))
|
|
channel_file = result.scalars().first()
|
|
if not channel_file:
|
|
return False
|
|
|
|
channel_file.message_id = message_id
|
|
channel_file.updated_at = int(time.time())
|
|
|
|
await db.commit()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
async def remove_file_from_channel_by_id(
|
|
self, channel_id: str, file_id: str, db: Optional[AsyncSession] = None
|
|
) -> bool:
|
|
try:
|
|
async with get_async_db_context(db) as db:
|
|
await db.execute(delete(ChannelFile).filter_by(channel_id=channel_id, file_id=file_id))
|
|
await db.commit()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
async def delete_channel_by_id(self, id: str, db: Optional[AsyncSession] = None) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
await AccessGrants.revoke_all_access('channel', id, db=db)
|
|
await db.execute(delete(Channel).filter(Channel.id == id))
|
|
await db.commit()
|
|
return True
|
|
|
|
####################
|
|
# Webhook Methods
|
|
####################
|
|
|
|
async def insert_webhook(
|
|
self,
|
|
channel_id: str,
|
|
user_id: str,
|
|
form_data: ChannelWebhookForm,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> Optional[ChannelWebhookModel]:
|
|
async with get_async_db_context(db) as db:
|
|
webhook = ChannelWebhookModel(
|
|
id=str(uuid.uuid4()),
|
|
channel_id=channel_id,
|
|
user_id=user_id,
|
|
name=form_data.name,
|
|
profile_image_url=form_data.profile_image_url,
|
|
token=secrets.token_urlsafe(32),
|
|
last_used_at=None,
|
|
created_at=int(time.time_ns()),
|
|
updated_at=int(time.time_ns()),
|
|
)
|
|
db.add(ChannelWebhook(**webhook.model_dump()))
|
|
await db.commit()
|
|
return webhook
|
|
|
|
async def get_webhooks_by_channel_id(
|
|
self, channel_id: str, db: Optional[AsyncSession] = None
|
|
) -> list[ChannelWebhookModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.channel_id == channel_id))
|
|
webhooks = result.scalars().all()
|
|
return [ChannelWebhookModel.model_validate(w) for w in webhooks]
|
|
|
|
async def get_webhook_by_id(
|
|
self, webhook_id: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelWebhookModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))
|
|
webhook = result.scalars().first()
|
|
return ChannelWebhookModel.model_validate(webhook) if webhook else None
|
|
|
|
async def get_webhook_by_id_and_token(
|
|
self, webhook_id: str, token: str, db: Optional[AsyncSession] = None
|
|
) -> Optional[ChannelWebhookModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(
|
|
select(ChannelWebhook).filter(
|
|
ChannelWebhook.id == webhook_id,
|
|
ChannelWebhook.token == token,
|
|
)
|
|
)
|
|
webhook = result.scalars().first()
|
|
return ChannelWebhookModel.model_validate(webhook) if webhook else None
|
|
|
|
async def update_webhook_by_id(
|
|
self,
|
|
webhook_id: str,
|
|
form_data: ChannelWebhookForm,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> Optional[ChannelWebhookModel]:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))
|
|
webhook = result.scalars().first()
|
|
if not webhook:
|
|
return None
|
|
webhook.name = form_data.name
|
|
webhook.profile_image_url = form_data.profile_image_url
|
|
webhook.updated_at = int(time.time_ns())
|
|
await db.commit()
|
|
return ChannelWebhookModel.model_validate(webhook)
|
|
|
|
async def update_webhook_last_used_at(self, webhook_id: str, db: Optional[AsyncSession] = None) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))
|
|
webhook = result.scalars().first()
|
|
if not webhook:
|
|
return False
|
|
webhook.last_used_at = int(time.time_ns())
|
|
await db.commit()
|
|
return True
|
|
|
|
async def delete_webhook_by_id(self, webhook_id: str, db: Optional[AsyncSession] = None) -> bool:
|
|
async with get_async_db_context(db) as db:
|
|
result = await db.execute(delete(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))
|
|
await db.commit()
|
|
return result.rowcount > 0
|
|
|
|
|
|
Channels = ChannelTable()
|