"""PR Preview Environments — data model. When a Git pull request is opened/updated, ServerKit deploys an isolated preview of that branch to a temporary domain and tears it down when the PR closes. Preview config lives here (its own tables) rather than on Application so the application schema stays untouched. Two tables: * ``application_previews`` — one row per live PR preview. * ``application_preview_settings`` — per-application opt-in + domain template. """ from datetime import datetime from app import db from app.models.mixins import TimestampMixin from app.models.json_column_mixin import JsonColumnMixin class ApplicationPreview(TimestampMixin, JsonColumnMixin, db.Model): """A single PR preview environment for an application.""" __tablename__ = 'application_previews' # Status lifecycle (string column, not an enum, to match the rest of the # codebase's loose status columns): STATUS_QUEUED = 'queued' STATUS_BUILDING = 'building' STATUS_RUNNING = 'running' STATUS_STOPPED = 'stopped' STATUS_FAILED = 'failed' STATUS_UNAVAILABLE = 'unavailable' STATUS_DESTROYED = 'destroyed' id = db.Column(db.Integer, primary_key=True) application_id = db.Column( db.Integer, db.ForeignKey('applications.id'), nullable=False, index=True) # Parent-side backref so the delete-cascade policy covers the NOT NULL FK. application = db.relationship('Application', backref=db.backref('previews', lazy='dynamic')) pr_number = db.Column(db.Integer, nullable=False) pr_title = db.Column(db.String(500), nullable=True) branch = db.Column(db.String(255), nullable=True) status = db.Column(db.String(20), default=STATUS_QUEUED, nullable=False) domain = db.Column(db.String(255), nullable=True) # JSON-encoded list of provisioned container ids (best-effort; may be empty # when Docker isn't available, e.g. dev/test). container_ids = db.Column(db.Text, nullable=True) # Durable intent ledger: written before external effects, retained on cleanup failure. request_token = db.Column(db.String(32), nullable=True) requested_action = db.Column(db.String(16), nullable=True) runtime_state = db.Column(db.Text, nullable=True) error_message = db.Column(db.Text, nullable=True) commit_sha = db.Column(db.String(64), nullable=True) expires_at = db.Column(db.DateTime, nullable=True) deleted_at = db.Column(db.DateTime, nullable=True) def get_container_ids(self): """The provisioned container ids as a list (empty when unset/invalid).""" return self._json_read('container_ids', [], expect=list) def set_container_ids(self, ids): self._json_write('container_ids', list(ids or []), falsy='[]') def get_runtime_state(self): return self._json_read('runtime_state', {}, expect=dict) def set_runtime_state(self, state): self._json_write('runtime_state', state, falsy='{}') def to_dict(self): container_ids = self.get_container_ids() # Older generic previews were marked running after config generation # alone. Do not advertise those rows as a deployed runtime on reads. status = self.status if status == self.STATUS_RUNNING and not container_ids: status = self.STATUS_UNAVAILABLE return { 'id': self.id, 'application_id': self.application_id, 'pr_number': self.pr_number, 'pr_title': self.pr_title, 'branch': self.branch, 'status': status, 'status_reason': ( self.error_message or ('No provisioned runtime is recorded for this preview.' if status == self.STATUS_UNAVAILABLE else None) ), 'domain': self.domain, 'url': (f"{self.get_runtime_state().get('scheme', 'https')}://{self.domain}" if self.domain and status == self.STATUS_RUNNING and container_ids else None), 'container_ids': container_ids, 'commit_sha': self.commit_sha, 'short_sha': self.commit_sha[:7] if self.commit_sha else None, 'expires_at': self.expires_at.isoformat() if self.expires_at else None, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'deleted_at': self.deleted_at.isoformat() if self.deleted_at else None, } def __repr__(self): return f'' class ApplicationPreviewSettings(TimestampMixin, db.Model): """Per-application PR-preview configuration (opt-in + domain template).""" __tablename__ = 'application_preview_settings' DEFAULT_DOMAIN_TEMPLATE = 'pr-{pr_number}.{app_domain}' application_id = db.Column( db.Integer, db.ForeignKey('applications.id'), primary_key=True, unique=True, nullable=False) # Parent-side backref so the delete-cascade policy covers the NOT NULL FK. application = db.relationship('Application', backref=db.backref('preview_settings', uselist=False)) inherit_environment = db.Column(db.Boolean, nullable=False, default=False, server_default='0') enabled = db.Column(db.Boolean, default=False, nullable=False) domain_template = db.Column( db.String(255), default=DEFAULT_DOMAIN_TEMPLATE, nullable=False) # Optional: deploy previews onto a specific server in the fleet. target_server_id = db.Column(db.String(36), nullable=True) # Optional TTL: previews are auto-destroyed this many days after creation. ttl_days = db.Column(db.Integer, nullable=True) def to_dict(self): return { 'application_id': self.application_id, 'enabled': bool(self.enabled), 'inherit_environment': bool(self.inherit_environment), 'domain_template': self.domain_template or self.DEFAULT_DOMAIN_TEMPLATE, 'target_server_id': self.target_server_id, 'ttl_days': self.ttl_days, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, } def __repr__(self): return f''