Coverage for open_webui/models/channels.py: 36%

1import secrets

2import time

3import uuid

4from typing import Optional

6from open_webui.internal.db import Base, JSONField, get_async_db_context

7from open_webui.models.access_grants import (

8 AccessGrantModel,

9 AccessGrants,

10)

11from open_webui.models.groups import Groups

12from open_webui.models.users import User

13from open_webui.utils.validate import validate_image_url

14from pydantic import BaseModel, ConfigDict, Field, field_validator

15from sqlalchemy import (

16 JSON,

17 BigInteger,

18 Boolean,

19 Column,

20 ForeignKey,

21 String,

22 Text,

23 UniqueConstraint,

24 and_,

25 case,

26 delete,

27 func,

28 or_,

29 select,

30 update,

31)

32from sqlalchemy.dialects.postgresql import JSONB

33from sqlalchemy.ext.asyncio import AsyncSession

35####################

36# Channel DB Schema

37####################

40class Channel(Base):

41 __tablename__ = 'channel'

43 id = Column(Text, primary_key=True, unique=True)

44 user_id = Column(Text)

45 type = Column(Text, nullable=True)

47 name = Column(Text)

48 description = Column(Text, nullable=True)

50 # Used to indicate if the channel is private (for 'group' type channels)

51 is_private = Column(Boolean, nullable=True)

53 data = Column(JSON, nullable=True)

54 meta = Column(JSON, nullable=True)

56 created_at = Column(BigInteger)

58 updated_at = Column(BigInteger)

59 updated_by = Column(Text, nullable=True)

61 archived_at = Column(BigInteger, nullable=True)

62 archived_by = Column(Text, nullable=True)

64 deleted_at = Column(BigInteger, nullable=True)

65 deleted_by = Column(Text, nullable=True)

68class ChannelModel(BaseModel):

69 model_config = ConfigDict(from_attributes=True)

71 id: str

72 user_id: str

74 type: Optional[str] = None

76 name: str

77 description: Optional[str] = None

79 is_private: Optional[bool] = None

81 data: Optional[dict] = None

82 meta: Optional[dict] = None

83 access_grants: list[AccessGrantModel] = Field(default_factory=list)

85 created_at: int # timestamp in epoch (time_ns)

87 updated_at: int # timestamp in epoch (time_ns)

88 updated_by: Optional[str] = None

90 archived_at: Optional[int] = None # timestamp in epoch (time_ns)

91 archived_by: Optional[str] = None

93 deleted_at: Optional[int] = None # timestamp in epoch (time_ns)

94 deleted_by: Optional[str] = None

97class ChannelMember(Base):

98 __tablename__ = 'channel_member'

100 id = Column(Text, primary_key=True, unique=True)

101 channel_id = Column(Text, nullable=False)

102 user_id = Column(Text, nullable=False)

103

104 role = Column(Text, nullable=True)

105 status = Column(Text, nullable=True)

106

107 is_active = Column(Boolean, nullable=False, default=True)

108

109 is_channel_muted = Column(Boolean, nullable=False, default=False)

110 is_channel_pinned = Column(Boolean, nullable=False, default=False)

111

112 data = Column(JSON, nullable=True)

113 meta = Column(JSON, nullable=True)

114

115 invited_at = Column(BigInteger, nullable=True)

116 invited_by = Column(Text, nullable=True)

117

118 joined_at = Column(BigInteger)

119 left_at = Column(BigInteger, nullable=True)

120

121 last_read_at = Column(BigInteger, nullable=True)

122

123 created_at = Column(BigInteger)

124 updated_at = Column(BigInteger)

125

126

127class ChannelMemberModel(BaseModel):

128 model_config = ConfigDict(from_attributes=True)

129

130 id: str

131 channel_id: str

132 user_id: str

133

134 role: Optional[str] = None

135 status: Optional[str] = None

136

137 is_active: bool = True

138

139 is_channel_muted: bool = False

140 is_channel_pinned: bool = False

141

142 data: Optional[dict] = None

143 meta: Optional[dict] = None

144

145 invited_at: Optional[int] = None # timestamp in epoch (time_ns)

146 invited_by: Optional[str] = None

147

148 joined_at: Optional[int] = None # timestamp in epoch (time_ns)

149 left_at: Optional[int] = None # timestamp in epoch (time_ns)

150

151 last_read_at: Optional[int] = None # timestamp in epoch (time_ns)

152

153 created_at: Optional[int] = None # timestamp in epoch (time_ns)

154 updated_at: Optional[int] = None # timestamp in epoch (time_ns)

155

156

157class ChannelFile(Base):

158 __tablename__ = 'channel_file'

159

160 id = Column(Text, unique=True, primary_key=True)

161 user_id = Column(Text, nullable=False)

162

163 channel_id = Column(Text, ForeignKey('channel.id', ondelete='CASCADE'), nullable=False)

164 message_id = Column(Text, ForeignKey('message.id', ondelete='CASCADE'), nullable=True)

165 file_id = Column(Text, ForeignKey('file.id', ondelete='CASCADE'), nullable=False)

166

167 created_at = Column(BigInteger, nullable=False)

168 updated_at = Column(BigInteger, nullable=False)

169

170 __table_args__ = (UniqueConstraint('channel_id', 'file_id', name='uq_channel_file_channel_file'),)

171

172

173class ChannelFileModel(BaseModel):

174 model_config = ConfigDict(from_attributes=True)

175

176 id: str

177

178 channel_id: str

179 file_id: str

180 user_id: str

181

182 created_at: int # timestamp in epoch (time_ns)

183 updated_at: int # timestamp in epoch (time_ns)

184

185

186class ChannelWebhook(Base):

187 __tablename__ = 'channel_webhook'

188

189 id = Column(Text, primary_key=True, unique=True)

190 channel_id = Column(Text, nullable=False)

191 user_id = Column(Text, nullable=False)

192

193 name = Column(Text, nullable=False)

194 profile_image_url = Column(Text, nullable=True)

195

196 token = Column(Text, nullable=False)

197 last_used_at = Column(BigInteger, nullable=True)

198

199 created_at = Column(BigInteger, nullable=False)

200 updated_at = Column(BigInteger, nullable=False)

201

202

203class ChannelWebhookModel(BaseModel):

204 model_config = ConfigDict(from_attributes=True)

205

206 id: str

207 channel_id: str

208 user_id: str

209

210 name: str

211 profile_image_url: Optional[str] = None

212

213 token: str

214 last_used_at: Optional[int] = None # timestamp in epoch (time_ns)

215

216 created_at: int # timestamp in epoch (time_ns)

217 updated_at: int # timestamp in epoch (time_ns)

218

219

220####################

221# Forms

222####################

223

224

225class ChannelResponse(ChannelModel):

226 is_manager: bool = False

227 write_access: bool = False

228

229 user_count: Optional[int] = None

230

231

232class ChannelForm(BaseModel):

233 name: str = ''

234 description: Optional[str] = None

235 is_private: Optional[bool] = None

236 data: Optional[dict] = None

237 meta: Optional[dict] = None

238 access_grants: Optional[list[dict]] = None

239 group_ids: Optional[list[str]] = None

240 user_ids: Optional[list[str]] = None

241

242

243class CreateChannelForm(ChannelForm):

244 type: Optional[str] = None

245

246

247class ChannelWebhookForm(BaseModel):

248 name: str

249 profile_image_url: Optional[str] = None

250

251 @field_validator('profile_image_url', mode='before')

252 @classmethod

253 def check_profile_image_url(cls, v: Optional[str]) -> Optional[str]:

254 if v is None:

255 return v

256 return validate_image_url(v)

257

258

259class ChannelTable:

260 async def _get_access_grants(self, channel_id: str, db: Optional[AsyncSession] = None) -> list[AccessGrantModel]:

261 return await AccessGrants.get_grants_by_resource('channel', channel_id, db=db)

262

263 async def _to_channel_model(

264 self,

265 channel: Channel,

266 access_grants: Optional[list[AccessGrantModel]] = None,

267 db: Optional[AsyncSession] = None,

268 ) -> ChannelModel:

269 channel_model = ChannelModel.model_validate(channel)

270 channel_model.access_grants = (

271 access_grants if access_grants is not None else await self._get_access_grants(channel_model.id, db=db)

272 )

273 return channel_model

274

275 async def _collect_unique_user_ids(

276 self,

277 invited_by: str,

278 user_ids: Optional[list[str]] = None,

279 group_ids: Optional[list[str]] = None,

280 ) -> set[str]:

281 """

282 Collect unique user ids from:

283 - invited_by

284 - user_ids

285 - each group in group_ids

286 Returns a set for efficient SQL diffing.

287 """

288 users = set(user_ids or [])

289 users.add(invited_by)

290

291 for group_id in group_ids or []:

292 group_user_ids = await Groups.get_group_user_ids_by_id(group_id)

293 users.update(group_user_ids)

294

295 return users

296

297 def _create_membership_models(

298 self,

299 channel_id: str,

300 invited_by: str,

301 user_ids: set[str],

302 ) -> list[ChannelMember]:

303 """

304 Takes a set of NEW user IDs (already filtered to exclude existing members).

305 Returns ORM ChannelMember objects to be added.

306 """

307 now = int(time.time_ns())

308 memberships = []

309

310 for uid in user_ids:

311 model = ChannelMemberModel(

312 **{

313 'id': str(uuid.uuid4()),

314 'channel_id': channel_id,

315 'user_id': uid,

316 'status': 'joined',

317 'is_active': True,

318 'is_channel_muted': False,

319 'is_channel_pinned': False,

320 'invited_at': now,

321 'invited_by': invited_by,

322 'joined_at': now,

323 'left_at': None,

324 'last_read_at': now,

325 'created_at': now,

326 'updated_at': now,

327 }

328 )

329 memberships.append(ChannelMember(**model.model_dump()))

330

331 return memberships

332

333 def _has_permission(self, db, query, filter: dict, permission: str = 'read'):

334 return AccessGrants.has_permission_filter(

335 db=db,

336 query=query,

337 DocumentModel=Channel,

338 filter=filter,

339 resource_type='channel',

340 permission=permission,

341 )

342

343 async def insert_new_channel(

344 self, form_data: CreateChannelForm, user_id: str, db: Optional[AsyncSession] = None

345 ) -> Optional[ChannelModel]:

346 async with get_async_db_context(db) as db:

347 channel = ChannelModel(

348 **{

349 **form_data.model_dump(exclude={'access_grants'}),

350 'type': form_data.type if form_data.type else None,

351 'name': form_data.name.lower(),

352 'id': str(uuid.uuid4()),

353 'user_id': user_id,

354 'created_at': int(time.time_ns()),

355 'updated_at': int(time.time_ns()),

356 'access_grants': [],

357 }

358 )

359 new_channel = Channel(**channel.model_dump(exclude={'access_grants'}))

360

361 if form_data.type in ['group', 'dm']:

362 users = await self._collect_unique_user_ids(

363 invited_by=user_id,

364 user_ids=form_data.user_ids,

365 group_ids=form_data.group_ids,

366 )

367 memberships = self._create_membership_models(

368 channel_id=new_channel.id,

369 invited_by=user_id,

370 user_ids=users,

371 )

372

373 db.add_all(memberships)

374 db.add(new_channel)

375 await db.commit()

376 await AccessGrants.set_access_grants('channel', new_channel.id, form_data.access_grants, db=db)

377 return await self._to_channel_model(new_channel, db=db)

378

379 async def get_channels(self, db: Optional[AsyncSession] = None) -> list[ChannelModel]:

380 async with get_async_db_context(db) as db:

381 result = await db.execute(select(Channel))

382 channels = result.scalars().all()

383 channel_ids = [channel.id for channel in channels]

384 grants_map = await AccessGrants.get_grants_by_resources('channel', channel_ids, db=db)

385 return [

386 await self._to_channel_model(

387 channel,

388 access_grants=grants_map.get(channel.id, []),

389 db=db,

390 )

391 for channel in channels

392 ]

393

394 async def get_channels_by_user_id(self, user_id: str, db: Optional[AsyncSession] = None) -> list[ChannelModel]:

395 async with get_async_db_context(db) as db:

396 user_group_ids = [group.id for group in await Groups.get_groups_by_member_id(user_id, db=db)]

397

398 result = await db.execute(

399 select(Channel)

400 .join(ChannelMember, Channel.id == ChannelMember.channel_id)

401 .filter(

402 Channel.deleted_at.is_(None),

403 Channel.archived_at.is_(None),

404 Channel.type.in_(['group', 'dm']),

405 ChannelMember.user_id == user_id,

406 ChannelMember.is_active.is_(True),

407 )

408 )

409 membership_channels = result.scalars().all()

410

411 stmt = select(Channel).filter(

412 Channel.deleted_at.is_(None),

413 Channel.archived_at.is_(None),

414 or_(

415 Channel.type.is_(None), # True NULL/None

416 Channel.type == '', # Empty string

417 and_(Channel.type != 'group', Channel.type != 'dm'),

418 ),

419 )

420 stmt = self._has_permission(db, stmt, {'user_id': user_id, 'group_ids': user_group_ids})

421

422 result = await db.execute(stmt)

423 standard_channels = result.scalars().all()

424

425 all_channels = list(membership_channels) + list(standard_channels)

426 channel_ids = [c.id for c in all_channels]

427 grants_map = await AccessGrants.get_grants_by_resources('channel', channel_ids, db=db)

428 return [

429 await self._to_channel_model(c, access_grants=grants_map.get(c.id, []), db=db) for c in all_channels

430 ]

431

432 async def get_dm_channel_by_user_ids(

433 self, user_ids: list[str], db: Optional[AsyncSession] = None

434 ) -> Optional[ChannelModel]:

435 async with get_async_db_context(db) as db:

436 # Ensure uniqueness in case a list with duplicates is passed

437 unique_user_ids = list(set(user_ids))

438

439 match_count = func.sum(

440 case(

441 (User.id.in_(unique_user_ids), 1),

442 else_=0,

443 )

444 )

445

446 subquery = (

447 select(ChannelMember.channel_id)

448 .join(User, User.id == ChannelMember.user_id)

449 .group_by(ChannelMember.channel_id)

450 # Match the exact set of accounts that still exist.

451 .having(func.count(User.id) == len(unique_user_ids))

452 .having(match_count == len(unique_user_ids))

453 .subquery()

454 )

455

456 result = await db.execute(

457 select(Channel)

458 .filter(

459 Channel.id.in_(select(subquery.c.channel_id)),

460 Channel.type == 'dm',

461 )

462 .limit(1)

463 )

464 channel = result.scalars().first()

465

466 return await self._to_channel_model(channel, db=db) if channel else None

467

468 async def add_members_to_channel(

469 self,

470 channel_id: str,

471 invited_by: str,

472 user_ids: Optional[list[str]] = None,

473 group_ids: Optional[list[str]] = None,

474 db: Optional[AsyncSession] = None,

475 ) -> list[ChannelMemberModel]:

476 async with get_async_db_context(db) as db:

477 # 1. Collect all user_ids including groups + inviter

478 requested_users = await self._collect_unique_user_ids(invited_by, user_ids, group_ids)

479

480 result = await db.execute(select(ChannelMember.user_id).filter(ChannelMember.channel_id == channel_id))

481 existing_users = {row[0] for row in result.all()}

482

483 new_user_ids = requested_users - existing_users

484 if not new_user_ids:

485 return [] # Nothing to add

486

487 new_memberships = self._create_membership_models(channel_id, invited_by, new_user_ids)

488

489 db.add_all(new_memberships)

490 await db.commit()

491

492 return [ChannelMemberModel.model_validate(membership) for membership in new_memberships]

493

494 async def remove_members_from_channel(

495 self,

496 channel_id: str,

497 user_ids: list[str],

498 db: Optional[AsyncSession] = None,

499 ) -> int:

500 async with get_async_db_context(db) as db:

501 result = await db.execute(

502 delete(ChannelMember).filter(

503 ChannelMember.channel_id == channel_id,

504 ChannelMember.user_id.in_(user_ids),

505 )

506 )

507 await db.commit()

508 return result.rowcount # number of rows deleted

509

510 async def is_user_channel_manager(self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None) -> bool:

511 async with get_async_db_context(db) as db:

512 result = await db.execute(select(Channel).filter(Channel.id == channel_id))

513 channel = result.scalars().first()

514 if channel and channel.user_id == user_id:

515 return True

516

517 result = await db.execute(

518 select(ChannelMember).filter(

519 ChannelMember.channel_id == channel_id,

520 ChannelMember.user_id == user_id,

521 ChannelMember.is_active.is_(True),

522 ChannelMember.role == 'manager',

523 )

524 )

525 membership = result.scalars().first()

526 return membership is not None

527

528 async def join_channel(

529 self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None

530 ) -> Optional[ChannelMemberModel]:

531 async with get_async_db_context(db) as db:

532 # Check if the membership already exists

533 result = await db.execute(

534 select(ChannelMember).filter(

535 ChannelMember.channel_id == channel_id,

536 ChannelMember.user_id == user_id,

537 )

538 )

539 existing_membership = result.scalars().first()

540 if existing_membership:

541 return ChannelMemberModel.model_validate(existing_membership)

542

543 # Create new membership

544 channel_member = ChannelMemberModel(

545 **{

546 'id': str(uuid.uuid4()),

547 'channel_id': channel_id,

548 'user_id': user_id,

549 'status': 'joined',

550 'is_active': True,

551 'is_channel_muted': False,

552 'is_channel_pinned': False,

553 'joined_at': int(time.time_ns()),

554 'left_at': None,

555 'last_read_at': int(time.time_ns()),

556 'created_at': int(time.time_ns()),

557 'updated_at': int(time.time_ns()),

558 }

559 )

560 new_membership = ChannelMember(**channel_member.model_dump())

561

562 db.add(new_membership)

563 await db.commit()

564 return channel_member

565

566 async def leave_channel(self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None) -> bool:

567 async with get_async_db_context(db) as db:

568 result = await db.execute(

569 select(ChannelMember).filter(

570 ChannelMember.channel_id == channel_id,

571 ChannelMember.user_id == user_id,

572 )

573 )

574 membership = result.scalars().first()

575 if not membership:

576 return False

577

578 membership.status = 'left'

579 membership.is_active = False

580 membership.left_at = int(time.time_ns())

581 membership.updated_at = int(time.time_ns())

582

583 await db.commit()

584 return True

585

586 async def get_member_by_channel_and_user_id(

587 self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None

588 ) -> Optional[ChannelMemberModel]:

589 async with get_async_db_context(db) as db:

590 result = await db.execute(

591 select(ChannelMember).filter(

592 ChannelMember.channel_id == channel_id,

593 ChannelMember.user_id == user_id,

594 )

595 )

596 membership = result.scalars().first()

597 return ChannelMemberModel.model_validate(membership) if membership else None

598

599 async def get_members_by_channel_id(

600 self, channel_id: str, db: Optional[AsyncSession] = None

601 ) -> list[ChannelMemberModel]:

602 async with get_async_db_context(db) as db:

603 result = await db.execute(select(ChannelMember).filter(ChannelMember.channel_id == channel_id))

604 memberships = result.scalars().all()

605 return [ChannelMemberModel.model_validate(membership) for membership in memberships]

606

607 async def pin_channel(

608 self,

609 channel_id: str,

610 user_id: str,

611 is_pinned: bool,

612 db: Optional[AsyncSession] = None,

613 ) -> bool:

614 async with get_async_db_context(db) as db:

615 result = await db.execute(

616 select(ChannelMember).filter(

617 ChannelMember.channel_id == channel_id,

618 ChannelMember.user_id == user_id,

619 )

620 )

621 membership = result.scalars().first()

622 if not membership:

623 return False

624

625 membership.is_channel_pinned = is_pinned

626 membership.updated_at = int(time.time_ns())

627

628 await db.commit()

629 return True

630

631 async def update_member_last_read_at(

632 self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None

633 ) -> bool:

634 async with get_async_db_context(db) as db:

635 result = await db.execute(

636 select(ChannelMember).filter(

637 ChannelMember.channel_id == channel_id,

638 ChannelMember.user_id == user_id,

639 )

640 )

641 membership = result.scalars().first()

642 if not membership:

643 return False

644

645 membership.last_read_at = int(time.time_ns())

646 membership.updated_at = int(time.time_ns())

647

648 await db.commit()

649 return True

650

651 async def update_member_active_status(

652 self,

653 channel_id: str,

654 user_id: str,

655 is_active: bool,

656 db: Optional[AsyncSession] = None,

657 ) -> bool:

658 async with get_async_db_context(db) as db:

659 result = await db.execute(

660 select(ChannelMember).filter(

661 ChannelMember.channel_id == channel_id,

662 ChannelMember.user_id == user_id,

663 )

664 )

665 membership = result.scalars().first()

666 if not membership:

667 return False

668

669 membership.is_active = is_active

670 membership.updated_at = int(time.time_ns())

671

672 await db.commit()

673 return True

674

675 async def is_user_channel_member(self, channel_id: str, user_id: str, db: Optional[AsyncSession] = None) -> bool:

676 async with get_async_db_context(db) as db:

677 result = await db.execute(

678 select(ChannelMember)

679 .filter(

680 ChannelMember.channel_id == channel_id,

681 ChannelMember.user_id == user_id,

682 ChannelMember.is_active.is_(True),

683 )

684 .limit(1)

685 )

686 membership = result.scalars().first()

687 return membership is not None

688

689 async def get_channel_by_id(self, id: str, db: Optional[AsyncSession] = None) -> Optional[ChannelModel]:

690 try:

691 async with get_async_db_context(db) as db:

692 result = await db.execute(select(Channel).filter(Channel.id == id))

693 channel = result.scalars().first()

694 return await self._to_channel_model(channel, db=db) if channel else None

695 except Exception:

696 return None

697

698 async def get_channels_by_file_id(self, file_id: str, db: Optional[AsyncSession] = None) -> list[ChannelModel]:

699 async with get_async_db_context(db) as db:

700 result = await db.execute(select(ChannelFile).filter(ChannelFile.file_id == file_id))

701 channel_files = result.scalars().all()

702 channel_ids = [cf.channel_id for cf in channel_files]

703 result = await db.execute(select(Channel).filter(Channel.id.in_(channel_ids)))

704 channels = result.scalars().all()

705 grants_map = await AccessGrants.get_grants_by_resources('channel', channel_ids, db=db)

706 return [

707 await self._to_channel_model(

708 channel,

709 access_grants=grants_map.get(channel.id, []),

710 db=db,

711 )

712 for channel in channels

713 ]

714

715 async def get_channels_by_file_id_and_user_id(

716 self, file_id: str, user_id: str, db: Optional[AsyncSession] = None

717 ) -> list[ChannelModel]:

718 async with get_async_db_context(db) as db:

719 # 1. Determine which channels have this file

720 result = await db.execute(select(ChannelFile).filter(ChannelFile.file_id == file_id))

721 channel_file_rows = result.scalars().all()

722 channel_ids = [row.channel_id for row in channel_file_rows]

723

724 if not channel_ids:

725 return []

726

727 # 2. Load all channel rows that still exist

728 result = await db.execute(

729 select(Channel).filter(

730 Channel.id.in_(channel_ids),

731 Channel.deleted_at.is_(None),

732 Channel.archived_at.is_(None),

733 )

734 )

735 channels = result.scalars().all()

736 if not channels:

737 return []

738

739 # Preload user's group membership

740 user_group_ids = [g.id for g in await Groups.get_groups_by_member_id(user_id, db=db)]

741

742 allowed_channels = []

743

744 for channel in channels:

745 # --- Case A: group or dm => user must be an active member ---

746 if channel.type in ['group', 'dm']:

747 result = await db.execute(

748 select(ChannelMember)

749 .filter(

750 ChannelMember.channel_id == channel.id,

751 ChannelMember.user_id == user_id,

752 ChannelMember.is_active.is_(True),

753 )

754 .limit(1)

755 )

756 membership = result.scalars().first()

757 if membership:

758 allowed_channels.append(await self._to_channel_model(channel, db=db))

759 continue

760

761 # --- Case B: standard channel => rely on ACL permissions ---

762 stmt = select(Channel).filter(Channel.id == channel.id)

763

764 stmt = self._has_permission(

765 db,

766 stmt,

767 {'user_id': user_id, 'group_ids': user_group_ids},

768 permission='read',

769 )

770

771 result = await db.execute(stmt)

772 allowed = result.scalars().first()

773 if allowed:

774 allowed_channels.append(await self._to_channel_model(allowed, db=db))

775

776 return allowed_channels

777

778 async def get_channel_by_id_and_user_id(

779 self, id: str, user_id: str, db: Optional[AsyncSession] = None

780 ) -> Optional[ChannelModel]:

781 async with get_async_db_context(db) as db:

782 # Fetch the channel

783 result = await db.execute(

784 select(Channel).filter(

785 Channel.id == id,

786 Channel.deleted_at.is_(None),

787 Channel.archived_at.is_(None),

788 )

789 )

790 channel = result.scalars().first()

791

792 if not channel:

793 return None

794

795 # If the channel is a group or dm, read access requires membership (active)

796 if channel.type in ['group', 'dm']:

797 result = await db.execute(

798 select(ChannelMember)

799 .filter(

800 ChannelMember.channel_id == id,

801 ChannelMember.user_id == user_id,

802 ChannelMember.is_active.is_(True),

803 )

804 .limit(1)

805 )

806 membership = result.scalars().first()

807 if membership:

808 return await self._to_channel_model(channel, db=db)

809 else:

810 return None

811

812 # For channels that are NOT group/dm, fall back to ACL-based read access

813 stmt = select(Channel).filter(Channel.id == id)

814

815 # Determine user groups

816 user_group_ids = [group.id for group in await Groups.get_groups_by_member_id(user_id, db=db)]

817

818 # Apply ACL rules

819 stmt = self._has_permission(

820 db,

821 stmt,

822 {'user_id': user_id, 'group_ids': user_group_ids},

823 permission='read',

824 )

825

826 result = await db.execute(stmt)

827 channel_allowed = result.scalars().first()

828 return await self._to_channel_model(channel_allowed, db=db) if channel_allowed else None

829

830 async def update_channel_by_id(

831 self, id: str, form_data: ChannelForm, db: Optional[AsyncSession] = None

832 ) -> Optional[ChannelModel]:

833 async with get_async_db_context(db) as db:

834 result = await db.execute(select(Channel).filter(Channel.id == id))

835 channel = result.scalars().first()

836 if not channel:

837 return None

838

839 channel.name = form_data.name

840 channel.description = form_data.description

841 channel.is_private = form_data.is_private

842

843 channel.data = form_data.data

844 channel.meta = form_data.meta

845

846 if form_data.access_grants is not None:

847 await AccessGrants.set_access_grants('channel', id, form_data.access_grants, db=db)

848 channel.updated_at = int(time.time_ns())

849

850 await db.commit()

851 return await self._to_channel_model(channel, db=db) if channel else None

852

853 async def add_file_to_channel_by_id(

854 self, channel_id: str, file_id: str, user_id: str, db: Optional[AsyncSession] = None

855 ) -> Optional[ChannelFileModel]:

856 async with get_async_db_context(db) as db:

857 channel_file = ChannelFileModel(

858 **{

859 'id': str(uuid.uuid4()),

860 'channel_id': channel_id,

861 'file_id': file_id,

862 'user_id': user_id,

863 'created_at': int(time.time()),

864 'updated_at': int(time.time()),

865 }

866 )

867

868 try:

869 result = ChannelFile(**channel_file.model_dump())

870 db.add(result)

871 await db.commit()

872 if result:

873 return ChannelFileModel.model_validate(result)

874 else:

875 return None

876 except Exception:

877 return None

878

879 async def set_file_message_id_in_channel_by_id(

880 self,

881 channel_id: str,

882 file_id: str,

883 message_id: str,

884 db: Optional[AsyncSession] = None,

885 ) -> bool:

886 try:

887 async with get_async_db_context(db) as db:

888 result = await db.execute(select(ChannelFile).filter_by(channel_id=channel_id, file_id=file_id))

889 channel_file = result.scalars().first()

890 if not channel_file:

891 return False

892

893 channel_file.message_id = message_id

894 channel_file.updated_at = int(time.time())

895

896 await db.commit()

897 return True

898 except Exception:

899 return False

900

901 async def remove_file_from_channel_by_id(

902 self, channel_id: str, file_id: str, db: Optional[AsyncSession] = None

903 ) -> bool:

904 try:

905 async with get_async_db_context(db) as db:

906 await db.execute(delete(ChannelFile).filter_by(channel_id=channel_id, file_id=file_id))

907 await db.commit()

908 return True

909 except Exception:

910 return False

911

912 async def delete_channel_by_id(self, id: str, db: Optional[AsyncSession] = None) -> bool:

913 async with get_async_db_context(db) as db:

914 await AccessGrants.revoke_all_access('channel', id, db=db)

915 await db.execute(delete(Channel).filter(Channel.id == id))

916 await db.commit()

917 return True

918

919 ####################

920 # Webhook Methods

921 ####################

922

923 async def insert_webhook(

924 self,

925 channel_id: str,

926 user_id: str,

927 form_data: ChannelWebhookForm,

928 db: Optional[AsyncSession] = None,

929 ) -> Optional[ChannelWebhookModel]:

930 async with get_async_db_context(db) as db:

931 webhook = ChannelWebhookModel(

932 id=str(uuid.uuid4()),

933 channel_id=channel_id,

934 user_id=user_id,

935 name=form_data.name,

936 profile_image_url=form_data.profile_image_url,

937 token=secrets.token_urlsafe(32),

938 last_used_at=None,

939 created_at=int(time.time_ns()),

940 updated_at=int(time.time_ns()),

941 )

942 db.add(ChannelWebhook(**webhook.model_dump()))

943 await db.commit()

944 return webhook

945

946 async def get_webhooks_by_channel_id(

947 self, channel_id: str, db: Optional[AsyncSession] = None

948 ) -> list[ChannelWebhookModel]:

949 async with get_async_db_context(db) as db:

950 result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.channel_id == channel_id))

951 webhooks = result.scalars().all()

952 return [ChannelWebhookModel.model_validate(w) for w in webhooks]

953

954 async def get_webhook_by_id(

955 self, webhook_id: str, db: Optional[AsyncSession] = None

956 ) -> Optional[ChannelWebhookModel]:

957 async with get_async_db_context(db) as db:

958 result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))

959 webhook = result.scalars().first()

960 return ChannelWebhookModel.model_validate(webhook) if webhook else None

961

962 async def get_webhook_by_id_and_token(

963 self, webhook_id: str, token: str, db: Optional[AsyncSession] = None

964 ) -> Optional[ChannelWebhookModel]:

965 async with get_async_db_context(db) as db:

966 result = await db.execute(

967 select(ChannelWebhook).filter(

968 ChannelWebhook.id == webhook_id,

969 ChannelWebhook.token == token,

970 )

971 )

972 webhook = result.scalars().first()

973 return ChannelWebhookModel.model_validate(webhook) if webhook else None

974

975 async def update_webhook_by_id(

976 self,

977 webhook_id: str,

978 form_data: ChannelWebhookForm,

979 db: Optional[AsyncSession] = None,

980 ) -> Optional[ChannelWebhookModel]:

981 async with get_async_db_context(db) as db:

982 result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))

983 webhook = result.scalars().first()

984 if not webhook:

985 return None

986 webhook.name = form_data.name

987 webhook.profile_image_url = form_data.profile_image_url

988 webhook.updated_at = int(time.time_ns())

989 await db.commit()

990 return ChannelWebhookModel.model_validate(webhook)

991

992 async def update_webhook_last_used_at(self, webhook_id: str, db: Optional[AsyncSession] = None) -> bool:

993 async with get_async_db_context(db) as db:

994 result = await db.execute(select(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))

995 webhook = result.scalars().first()

996 if not webhook:

997 return False

998 webhook.last_used_at = int(time.time_ns())

999 await db.commit()

1000 return True

1001

1002 async def delete_webhook_by_id(self, webhook_id: str, db: Optional[AsyncSession] = None) -> bool:

1003 async with get_async_db_context(db) as db:

1004 result = await db.execute(delete(ChannelWebhook).filter(ChannelWebhook.id == webhook_id))

1005 await db.commit()

1006 return result.rowcount > 0

1007

1008

1009Channels = ChannelTable()