""" Copyright 2024-2026 ChatterMate Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. """ from sqlalchemy.orm import Session from app.models.widget_app import WidgetApp from app.core.security import ( generate_widget_api_key, hash_widget_api_key, verify_widget_api_key, widget_api_key_lookup, ) from typing import Optional, List from uuid import UUID from app.core.logger import get_logger logger = get_logger(__name__) class WidgetAppRepository: def __init__(self, db: Session): self.db = db def create_app( self, organization_id: UUID, created_by: UUID, name: str, description: Optional[str] = None ) -> tuple[WidgetApp, str]: """ Create a new widget app with generated API key. Returns: tuple: (WidgetApp object, plain_text_api_key) Plain key is returned ONCE and never stored. """ try: # Generate new API key plain_key = generate_widget_api_key() app = WidgetApp( name=name, description=description, organization_id=organization_id, created_by=created_by, api_key_hash=hash_widget_api_key(plain_key), api_key_lookup=widget_api_key_lookup(plain_key), is_active=True ) self.db.add(app) self.db.commit() self.db.refresh(app) logger.info(f"Created widget app: {app.id} for org: {organization_id}") # Return both app and plain key (shown only once) return app, plain_key except Exception as e: self.db.rollback() logger.error(f"Error creating widget app: {str(e)}") raise def get_app_by_id(self, app_id: UUID, organization_id: UUID) -> Optional[WidgetApp]: """Get widget app by ID (scoped to organization)""" return self.db.query(WidgetApp).filter( WidgetApp.id == app_id, WidgetApp.organization_id == organization_id ).first() def get_apps_by_organization( self, organization_id: UUID, include_inactive: bool = False ) -> List[WidgetApp]: """Get all widget apps for an organization""" query = self.db.query(WidgetApp).filter( WidgetApp.organization_id == organization_id ) if not include_inactive: query = query.filter(WidgetApp.is_active == True) return query.order_by(WidgetApp.created_at.desc()).all() def validate_api_key(self, api_key: str) -> Optional[WidgetApp]: """ Validate API key and return associated app if valid. One indexed query on the key's SHA-256. Keys issued before that column existed are bcrypt-checked against the apps that still have none (about 100ms each), then get their lookup filled in, so it happens once per key. Args: api_key: Plain text API key from Authorization header Returns: WidgetApp if valid, None otherwise """ lookup = widget_api_key_lookup(api_key) app = self.db.query(WidgetApp).filter( WidgetApp.api_key_lookup == lookup, WidgetApp.is_active == True ).first() if app: return app unindexed_apps = self.db.query(WidgetApp).filter( WidgetApp.api_key_lookup.is_(None), WidgetApp.is_active == True ).all() for app in unindexed_apps: if verify_widget_api_key(api_key, app.api_key_hash): app.api_key_lookup = lookup self.db.commit() logger.info(f"Valid API key for app: {app.id}; indexed its lookup") return app logger.warning("Invalid API key attempt") return None def update_app( self, app_id: UUID, organization_id: UUID, name: Optional[str] = None, description: Optional[str] = None, is_active: Optional[bool] = None ) -> Optional[WidgetApp]: """Update widget app (name, description, active status)""" app = self.get_app_by_id(app_id, organization_id) if not app: return None if name is not None: app.name = name if description is not None: app.description = description if is_active is not None: app.is_active = is_active self.db.commit() self.db.refresh(app) logger.info(f"Updated widget app: {app_id}") return app def deactivate_app(self, app_id: UUID, organization_id: UUID) -> bool: """Deactivate (soft delete) a widget app""" app = self.get_app_by_id(app_id, organization_id) if not app: return False app.is_active = False self.db.commit() logger.info(f"Deactivated widget app: {app_id}") return True def regenerate_api_key( self, app_id: UUID, organization_id: UUID ) -> Optional[tuple[WidgetApp, str]]: """ Regenerate API key for an existing app. Returns: tuple: (WidgetApp, new_plain_key) or None if app not found """ app = self.get_app_by_id(app_id, organization_id) if not app: return None # Generate new key new_plain_key = generate_widget_api_key() # The old key stops matching both ways at once app.api_key_hash = hash_widget_api_key(new_plain_key) app.api_key_lookup = widget_api_key_lookup(new_plain_key) self.db.commit() self.db.refresh(app) logger.info(f"Regenerated API key for app: {app_id}") return app, new_plain_key def delete_app(self, app_id: UUID, organization_id: UUID) -> bool: """ Hard delete a widget app (use with caution). Prefer deactivate_app() for soft delete. """ app = self.get_app_by_id(app_id, organization_id) if not app: return False self.db.delete(app) self.db.commit() logger.warning(f"HARD DELETED widget app: {app_id}") return True