""" This module defines Roundabout's SQLAlchemy database models. Roundabout - git hosting for everyone Copyright (C) 2023-2025 Roundabout developers This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see . """ __all__ = [ "RepoAccess", "RepoFavourite", "Repo", "UserFollow", "UserNotification", "User", "UserTrust", "Notification", "PostVote", "Post", "Commit", "PullRequest", "EmailChangeRequest", "Comment", "PullRequestResolvesThread", "Label", "PostLabel", ] import secrets import subprocess import markdown from app import app, db, bcrypt import git from datetime import datetime, timedelta from enum import Enum from PIL import Image from cairosvg import svg2png import os import config import cairosvg import random import celery_tasks with (app.app_context()): class RepoAccess(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) repo_route = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) access_level = db.Column(db.SmallInteger(), nullable=False) # 0 read-only, 1 read-write, 2 admin automatic = db.Column(db.Boolean, default=False, nullable=False, server_default="false") user = db.relationship("User", back_populates="repo_access") repo = db.relationship("Repo", back_populates="repo_access") __table_args__ = (db.UniqueConstraint("user_username", "repo_route", name="_user_repo_uc"),) def __init__(self, user, repo, level, automatic=False): self.user_username = user.username self.repo_route = repo.route self.access_level = level self.automatic = automatic class RepoFavourite(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) repo_route = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) notify_commit = db.Column(db.Boolean, default=False, nullable=False) notify_forum = db.Column(db.Boolean, default=False, nullable=False) notify_pr = db.Column(db.Boolean, default=False, nullable=False) notify_admin = db.Column(db.Boolean, default=False, nullable=False) user = db.relationship("User", back_populates="favourites") repo = db.relationship("Repo", back_populates="favourites") __table_args__ = (db.UniqueConstraint("user_username", "repo_route", name="_user_repo_uc1"),) def __init__(self, user, repo): self.user_username = user.username self.repo_route = repo.route class PostVote(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) post_identifier = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=False) vote_score = db.Column(db.SmallInteger(), nullable=False) user = db.relationship("User", back_populates="votes") post = db.relationship("Post", back_populates="votes") __table_args__ = (db.UniqueConstraint("user_username", "post_identifier", name="_user_post_uc"),) def __init__(self, user, post, score): self.user_username = user.username self.post_identifier = post.identifier self.vote_score = score class User(db.Model): username = db.Column(db.String(32), unique=True, nullable=False, primary_key=True) display_name = db.Column(db.Unicode(128), unique=False, nullable=True) bio = db.Column(db.Unicode(16384), unique=False, nullable=True) password_hashed = db.Column(db.String(60), nullable=False) email = db.Column(db.String(254), nullable=True) company = db.Column(db.Unicode(64), nullable=True) company_URL = db.Column(db.String(256), nullable=True) URL = db.Column(db.String(256), nullable=True) show_mail = db.Column(db.Boolean, default=False, nullable=False) location = db.Column(db.Unicode(64), nullable=True) creation_date = db.Column(db.DateTime, default=datetime.utcnow) default_page_length = db.Column(db.SmallInteger, nullable=False, default=32, server_default="32") max_post_nesting = db.Column(db.SmallInteger, nullable=False, default=3, server_default="3") repositories = db.relationship("Repo", back_populates="owner", cascade="all, delete-orphan") followers = db.relationship("UserFollow", back_populates="followed", foreign_keys="[UserFollow.followed_username]") follows = db.relationship("UserFollow", back_populates="follower", foreign_keys="[UserFollow.follower_username]") email_change_requests = db.relationship("EmailChangeRequest", back_populates="user") repo_access = db.relationship("RepoAccess", back_populates="user") votes = db.relationship("PostVote", back_populates="user") favourites = db.relationship("RepoFavourite", back_populates="user") pushes = db.relationship("Commit", back_populates="pusher", foreign_keys="[Commit.pusher_name]") posts = db.relationship("Post", back_populates="owner") comments = db.relationship("Comment", back_populates="owner") prs = db.relationship("PullRequest", back_populates="owner") notifications = db.relationship("UserNotification", back_populates="user") trusts = db.relationship("UserTrust", back_populates="host", foreign_keys="[UserTrust.host_username]") trusted_by = db.relationship("UserTrust", back_populates="trusted", foreign_keys="[UserTrust.trusted_username]") def __init__(self, username, password, email=None, display_name=None): self.username = username self.password_hashed = bcrypt.generate_password_hash(password, config.HASHING_ROUNDS).decode("utf-8") self.email = "" if email: email_change_request = EmailChangeRequest(self, email) db.session.add(email_change_request) db.session.flush() self.display_name = display_name # Create the user's directory if not os.path.exists(os.path.join(config.REPOS_PATH, username)): os.makedirs(os.path.join(config.REPOS_PATH, username)) if not os.path.exists(os.path.join(config.USERDATA_PATH, username)): os.makedirs(os.path.join(config.USERDATA_PATH, username)) avatar_name = random.choice(os.listdir(config.DEFAULT_AVATARS_PATH)) if os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name).endswith(".svg"): cairosvg.svg2png(url=os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name), write_to="/tmp/roundabout-avatar.png") avatar = Image.open("/tmp/roundabout-avatar.png") else: avatar = Image.open(os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name)) avatar.thumbnail(config.AVATAR_SIZE) avatar.save(os.path.join(config.USERDATA_PATH, username, "avatar.png")) # Create the configuration repo config_repo = Repo(self, ".config", 0) db.session.add(config_repo) notification = Notification({"type": "welcome"}) db.session.add(notification) db.session.commit() user_notification = UserNotification(self, notification, 1) db.session.add(user_notification) db.session.commit() celery_tasks.send_notification.apply_async(args=[user_notification.id]) class UserTrust(db.Model): id = db.Column(db.Integer, primary_key=True) host_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) trusted_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) trust_level = db.Column(db.SmallInteger, nullable=False) host = db.relationship("User", back_populates="trusts", foreign_keys=[host_username]) trusted = db.relationship("User", back_populates="trusted_by", foreign_keys=[trusted_username]) __table_args__ = (db.UniqueConstraint("host_username", "trusted_username", name="_host_trusted_uc"),) def __init__(self, host, trusted, level): self.host_username = host.username self.trusted_username = trusted.username self.trust_level = level # Add user to all of the host's repositories for repo in host.repositories: existing_relationship = RepoAccess.query.filter_by(user=trusted, repo=repo).first() if existing_relationship: continue relationship = RepoAccess(trusted, repo, level, automatic=True) db.session.add(relationship) def cancel(self): """Remove the trusted user from all of the host's repositories.""" relationships = RepoAccess.query.filter(RepoAccess.repo.has(owner_name=self.host_username), RepoAccess.user == self.trusted, RepoAccess.automatic == True) relationships.delete() db.session.delete(self) class Repo(db.Model): route = db.Column(db.String(98), unique=True, nullable=False, primary_key=True) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) name = db.Column(db.String(64), nullable=False) owner = db.relationship("User", back_populates="repositories") visibility = db.Column(db.SmallInteger(), nullable=False) info = db.Column(db.Unicode(512), nullable=True) url = db.Column(db.String(256), nullable=True) creation_date = db.Column(db.DateTime, default=datetime.utcnow) default_branch = db.Column(db.String(64), nullable=True, default="") commits = db.relationship("Commit", back_populates="repo", cascade="all, delete-orphan") posts = db.relationship("Post", back_populates="repo", cascade="all, delete-orphan") comments = db.relationship("Comment", back_populates="repo", cascade="all, delete-orphan") repo_access = db.relationship("RepoAccess", back_populates="repo", cascade="all, delete-orphan") favourites = db.relationship("RepoFavourite", back_populates="repo", cascade="all, delete-orphan") bases = db.relationship("PullRequest", back_populates="base", foreign_keys="[PullRequest.base_route]", cascade="all, delete-orphan") labels = db.relationship("Label", back_populates="repo", cascade="all, delete-orphan") has_site = db.Column(db.SmallInteger, nullable=False, default=0, # 0 means no site, 1 means it's got a site, 2 means it's the user's primary site server_default="0") # (the one accessible at username.localhost) site_branch = db.Column(db.String(64), nullable=True) last_post_id = db.Column(db.Integer, nullable=False, default=0, server_default="0") last_comment_id = db.Column(db.Integer, nullable=False, default=0, server_default="0") def __init__(self, owner, name, visibility): self.route = f"/{owner.username}/{name}" self.name = name self.owner_name = owner.username self.owner = owner self.visibility = visibility # Add the owner as an admin repo_access = RepoAccess(owner, self, 2) db.session.add(repo_access) with db.session.no_autoflush: # Add the trusted users to the repo for trust in owner.trusts: if trust.trust_level > 0: repo_access = RepoAccess(trust.trusted, self, trust.trust_level, automatic=True) db.session.add(repo_access) # Create the directory if not os.path.exists(os.path.join(config.REPOS_PATH, self.owner_name, self.name)): subprocess.run(["git", "init", self.name], cwd=os.path.join(config.REPOS_PATH, self.owner_name)) class Commit(db.Model): identifier = db.Column(db.String(227), unique=True, nullable=False, primary_key=True) sha = db.Column(db.String(128), nullable=False) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) owner_name = db.Column(db.String(128), nullable=False) owner_identity = db.Column(db.String(321)) pusher_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) receive_date = db.Column(db.DateTime, default=datetime.now) author_date = db.Column(db.DateTime) message = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="commits") pusher = db.relationship("User", back_populates="pushes", foreign_keys=[pusher_name]) comments = db.relationship("Comment", back_populates="commit") def __init__(self, sha, owner, repo, date, message, owner_identity, pusher, owner_name=None): self.identifier = f"{repo.route}/{sha}" self.sha = sha self.repo_name = repo.route self.repo = repo if not owner: self.owner_name = owner_name else: self.owner_name = owner.username self.pusher_name = pusher.username self.pusher = pusher self.author_date = datetime.fromtimestamp(int(date)) self.message = message self.owner_identity = owner_identity notification = Notification({"type": "commit", "repo": repo.route, "commit": sha}) db.session.add(notification) db.session.commit() # save the notification to get the ID # Send a notification to all users who have enabled commit notifications for this repo for relationship in RepoFavourite.query.filter_by(repo_route=repo.route, notify_commit=True).all(): user = relationship.user user_notification = UserNotification(user, notification, 1) db.session.add(user_notification) db.session.commit() celery_tasks.send_notification.apply_async(args=[user_notification.id]) class Label(db.Model): identifier = db.Column(db.String(162), unique=True, nullable=False, primary_key=True) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) name = db.Column(db.Unicode(64), nullable=False) colour = db.Column(db.Integer, nullable=False, server_default="0") repo = db.relationship("Repo", back_populates="labels") posts = db.relationship("PostLabel", back_populates="label") def __init__(self, repo, name, colour): self.identifier = f"{repo.route}/" + secrets.token_hex(32) # randomise label IDs self.name = name self.colour = int(colour.removeprefix("#"), 16) self.repo_name = repo.route @property def colour_hex(self): return f"#{self.colour:06x}" @colour_hex.setter def colour_hex(self, value): self.colour = int(value.removeprefix("#"), 16) class PostLabel(db.Model): id = db.Column(db.Integer, primary_key=True) post_identifier = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=False) label_identifier = db.Column(db.String(162), db.ForeignKey("label.identifier"), nullable=False) post = db.relationship("Post", back_populates="labels") label = db.relationship("Label", back_populates="posts") def __init__(self, post, label): self.post_identifier = post.identifier self.post = post self.label = label class Post(db.Model): identifier = db.Column(db.String(109), unique=True, nullable=False, primary_key=True) number = db.Column(db.Integer, nullable=False) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) votes = db.relationship("PostVote", back_populates="post") vote_sum = db.Column(db.Integer, nullable=False, default=0) parent_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=True) root_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=True) state = db.Column(db.SmallInteger, nullable=True, default=1) date = db.Column(db.DateTime, default=datetime.now) last_updated = db.Column(db.DateTime, default=datetime.now) subject = db.Column(db.Unicode(384)) message = db.Column(db.UnicodeText) html = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="posts") owner = db.relationship("User", back_populates="posts") parent = db.relationship("Post", back_populates="children", primaryjoin="Post.parent_id==Post.identifier", foreign_keys="[Post.parent_id]", remote_side="Post.identifier") root = db.relationship("Post", primaryjoin="Post.root_id==Post.identifier", foreign_keys="[Post.root_id]", remote_side="Post.identifier", post_update=True) children = db.relationship("Post", remote_side="Post.parent_id", primaryjoin="Post.identifier==Post.parent_id", foreign_keys="[Post.parent_id]") resolved_by = db.relationship("PullRequestResolvesThread", back_populates="post") labels = db.relationship("PostLabel", back_populates="post") def __init__(self, owner, repo, parent, subject, message): self.identifier = f"{repo.route}/{repo.last_post_id}" self.number = repo.last_post_id self.repo_name = repo.route self.repo = repo self.owner_name = owner.username self.owner = owner self.subject = subject self.message = message self.html = markdown.markdown2html(message).prettify() self.parent = parent if parent: self.root = parent.root else: self.root = self repo.last_post_id += 1 notification = Notification({"type": "post", "repo": repo.route, "post": self.identifier}) db.session.add(notification) db.session.commit() # save the notification to get the ID # Send a notification to all users who have enabled forum notifications for this repo for relationship in RepoFavourite.query.filter_by(repo_route=repo.route, notify_forum=True).all(): user = relationship.user user_notification = UserNotification(user, notification, 1) db.session.add(user_notification) db.session.commit() celery_tasks.send_notification.apply_async(args=[user_notification.id]) def update_date(self): self.last_updated = datetime.now() with db.session.no_autoflush: if self.parent is not None: self.parent.update_date() class Comment(db.Model): identifier = db.Column(db.String(109), unique=True, nullable=False, primary_key=True) number = db.Column(db.Integer, nullable=False) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) commit_identifier = db.Column(db.String(227), db.ForeignKey("commit.identifier"), nullable=False) pr_id = db.Column(db.BigInteger, db.ForeignKey("pull_request.id"), nullable=True) file = db.Column(db.String(256), nullable=True) line_number = db.Column(db.Integer, nullable=True) line_type = db.Column(db.SmallInteger, nullable=True, default=0, server_default="0") # 0 is deleted, 1 is modified state = db.Column(db.SmallInteger, nullable=True, default=1) review = db.Column(db.SmallInteger, nullable=True, default=0) date = db.Column(db.DateTime, default=datetime.now) message = db.Column(db.UnicodeText) html = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="comments") owner = db.relationship("User", back_populates="comments") commit = db.relationship("Commit", back_populates="comments") def __init__(self, owner, repo, commit, message, file, line_number, pr=None): self.identifier = f"{repo.route}/{repo.last_comment_id}" self.number = repo.last_comment_id self.repo_name = repo.route self.repo = repo self.owner_name = owner.username self.owner = owner self.commit_identifier = commit.identifier self.commit = commit self.message = message self.html = markdown.markdown2html(message).prettify() self.file = file self.line_number = int(line_number[1:]) self.line_type = int(line_number[0] == "+") if pr: self.pr = pr repo.last_comment_id += 1 @property def text(self): return self.html @text.setter def text(self, value): self.html = markdown.markdown2html(value).prettify() self.message = value # message is stored in markdown format for future editing or plaintext display class UserNotification(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) notification_id = db.Column(db.BigInteger, db.ForeignKey("notification.id")) attention_level = db.Column(db.SmallInteger, nullable=False) # 0 is read read_time = db.Column(db.DateTime, nullable=True) user = db.relationship("User", back_populates="notifications") notification = db.relationship("Notification", back_populates="notifications") __table_args__ = (db.UniqueConstraint("user_username", "notification_id", name="_user_notification_uc"),) def __init__(self, user, notification, level): self.user_username = user.username self.notification_id = notification.id self.attention_level = level def mark_read(self): self.read_time = datetime.utcnow() self.attention_level = 0 def mark_unread(self): self.attention_level = 4 class UserFollow(db.Model): id = db.Column(db.Integer, primary_key=True) follower_username = db.Column(db.String(32), db.ForeignKey("user.username", ondelete="CASCADE"), nullable=False) followed_username = db.Column(db.String(32), db.ForeignKey("user.username", ondelete="CASCADE"), nullable=False) follower = db.relationship("User", back_populates="followers", foreign_keys=[follower_username]) followed = db.relationship("User", back_populates="follows", foreign_keys=[followed_username]) def __init__(self, follower_username, followed_username): self.follower_username = follower_username self.followed_username = followed_username class Notification(db.Model): id = db.Column(db.BigInteger, primary_key=True, autoincrement=True) data = db.Column(db.dialects.postgresql.JSONB, nullable=False, default={}) notifications = db.relationship("UserNotification", back_populates="notification") timestamp = db.Column(db.DateTime, nullable=False, default=datetime.now) def __init__(self, json): self.data = json class PullRequestResolvesThread(db.Model): id = db.Column(db.Integer, primary_key=True) pr_id = db.Column(db.BigInteger, db.ForeignKey("pull_request.id"), nullable=False) post_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=False) pr = db.relationship("PullRequest", back_populates="resolves") post = db.relationship("Post", back_populates="resolved_by") def __init__(self, pr, post): self.pr = pr self.post = post class PullRequest(db.Model): id = db.Column(db.BigInteger, primary_key=True, autoincrement=True) head_route = db.Column(db.String(256), nullable=False) base_route = db.Column(db.String(98), db.ForeignKey("repo.route", ondelete="CASCADE"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) state = db.Column(db.SmallInteger, nullable=False, default=0) # 0 pending, 1 merged, 2 rejected base = db.relationship("Repo", back_populates="bases", foreign_keys=[base_route]) head_branch = db.Column(db.String(64), nullable=False) base_branch = db.Column(db.String(64), nullable=False) owner = db.relationship("User", back_populates="prs") resolves = db.relationship("PullRequestResolvesThread", back_populates="pr") timestamp = db.Column(db.DateTime, nullable=False, default=datetime.now) def __init__(self, head_route, head_branch, base, base_branch, owner): self.head_route = head_route self.base = base self.head_branch = head_branch self.base_branch = base_branch self.owner = owner @property def resolves_list(self): return " ".join([str(post.post.number) for post in self.resolves]) @resolves_list.setter def resolves_list(self, value): link_to = [Post.query.filter_by(number=int(number), repo=self.base).first() for number in value.split()] resolved_posts = [post.post for post in self.resolves] no_longer_resolves = list(set(resolved_posts) - set(link_to)) for post in no_longer_resolves: db.session.delete(PullRequestResolvesThread.query.filter_by(pr=self, post=post).first()) for post in link_to: if post not in resolved_posts and post is not None and not post.parent: # only top-level posts can be resolved db.session.add(PullRequestResolvesThread(self, post)) db.session.commit() class EmailChangeRequest(db.Model): id = db.Column(db.BigInteger, primary_key=True, autoincrement=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) new_email = db.Column(db.String(254), nullable=False) code = db.Column(db.String(64), nullable=False) expires_on = db.Column(db.DateTime, nullable=False) user = db.relationship("User", back_populates="email_change_requests") def __init__(self, user, new_email): self.user = user self.new_email = new_email self.code = hex(secrets.randbits(256)).removeprefix("0x") self.expires_on = datetime.now() + timedelta(days=1)