# services/forum.py
from snek.system.service import BaseService
import re
import uuid
from collections import defaultdict
from typing import Any, Awaitable, Callable, Dict, List
from snek.system.model import now
EventListener = Callable[[str, Any], Awaitable[None]] | Callable[[str, Any], None]
class BaseForumService(BaseService):
"""
Base mix-in that gives a service `add_notification_listener`,
an internal `_dispatch_event` helper, and a public `notify` method.
"""
def get_timestamp(self):
return now()
def generate_uid(self):
return str(uuid.uuid4())
def __init__(self, *args, **kwargs) -> None:
# Map event name -> list of listener callables
self._listeners: Dict[str, List[EventListener]] = defaultdict(list)
super().__init__(*args, **kwargs)
def add_notification_listener(
self, event_name: str, listener: EventListener
) -> None:
"""
Register a callback to be fired when `event_name` happens.
Parameters
----------
event_name : str
The name of the domain event, e.g. "post_created".
listener : Callable[[str, Any], Awaitable[None]] | Callable[[str, Any], None]
Your handler; can be async or sync.
"""
if not callable(listener):
raise TypeError("listener must be callable")
self._listeners[event_name].append(listener)
async def _dispatch_event(self, event_name: str, data: Any) -> None:
"""Invoke every listener for the given event."""
for listener in self._listeners.get(event_name, []):
if hasattr(listener, "__await__"): # async function or coro
await listener(event_name, data)
else: # plain sync function
listener(event_name, data)
async def notify(self, event_name: str, data: Any) -> None:
"""
Public method to trigger notification to all listeners of an event.
Parameters
----------
event_name : str
The name of the event to notify listeners about.
data : Any
The data to pass to the listeners.
"""
await self._dispatch_event(event_name, data)
class ForumService(BaseForumService):
mapper_name = "forum"
async def create_forum(self, name, description, created_by_uid, slug=None, icon=None):
if not slug:
slug = self.generate_slug(name)
# Check if slug exists
existing = await self.get(slug=slug)
if existing:
slug = f"{slug}-{self.generate_uid()[:8]}"
model = await self.new()
model["name"] = name
model["description"] = description
model["slug"] = slug
model["created_by_uid"] = created_by_uid
if icon:
model["icon"] = icon
if await self.save(model):
await self.notify("forum_created", model)
return model
raise Exception(f"Failed to create forum: {model.errors}")
def generate_slug(self, text):
# Convert to lowercase and replace spaces with hyphens
slug = text.lower().strip()
slug = re.sub(r'[^\w\s-]', '', slug)
slug = re.sub(r'[-\s]+', '-', slug)
return slug
async def get_active_forums(self):
async for forum in self.find(is_active=True, order_by="position"):
yield forum
async def update_last_post(self, forum_uid, thread_uid):
forum = await self.get(uid=forum_uid)
if forum:
forum["last_post_at"] = self.get_timestamp()
forum["last_thread_uid"] = thread_uid
await self.save(forum)
# services/thread.py
class ThreadService(BaseForumService):
mapper_name = "thread"
async def create_thread(self, forum_uid, title, content, created_by_uid):
# Generate slug
slug = self.services.forum.generate_slug(title)
# Check if slug exists in this forum
existing = await self.get(forum_uid=forum_uid, slug=slug)
if existing:
slug = f"{slug}-{self.generate_uid()[:8]}"
# Create thread
thread = await self.new()
thread["forum_uid"] = forum_uid
thread["title"] = title
thread["slug"] = slug
thread["created_by_uid"] = created_by_uid
thread["last_post_at"] = self.get_timestamp()
thread["last_post_by_uid"] = created_by_uid
if await self.save(thread):
# Create first post
post = await self.services.post.create_post(
thread_uid=thread["uid"],
forum_uid=forum_uid,
content=content,
created_by_uid=created_by_uid,
is_first_post=True
)
# Update forum counters
forum = await self.services.forum.get(uid=forum_uid)
await forum.increment_thread_count()
await self.services.forum.update_last_post(forum_uid, thread["uid"])
await self.notify("thread_created", {
"thread": thread,
"forum_uid": forum_uid
})
return thread, post
raise Exception(f"Failed to create thread: {thread.errors}")
async def toggle_pin(self, thread_uid, user_uid):
thread = await self.get(uid=thread_uid)
if not thread:
return None
# Check if user is admin
user = await self.services.user.get(uid=user_uid)
if not user.get("is_admin"):
return None
thread["is_pinned"] = not thread["is_pinned"]
await self.save(thread)
return thread
async def toggle_lock(self, thread_uid, user_uid):
thread = await self.get(uid=thread_uid)
if not thread:
return None
# Check if user is admin or thread creator
user = await self.services.user.get(uid=user_uid)
if not user.get("is_admin") and thread["created_by_uid"] != user_uid:
return None
thread["is_locked"] = not thread["is_locked"]
await self.save(thread)
return thread
# services/post.py
class PostService(BaseForumService):
mapper_name = "post"
async def create_post(self, thread_uid, forum_uid, content, created_by_uid, is_first_post=False):
# Check if thread is locked
thread = await self.services.thread.get(uid=thread_uid)
if thread["is_locked"] and not is_first_post:
raise Exception("Thread is locked")
post = await self.new()
post["thread_uid"] = thread_uid
post["forum_uid"] = forum_uid
post["content"] = content
post["created_by_uid"] = created_by_uid
post["is_first_post"] = is_first_post
if await self.save(post):
# Update thread counters
if not is_first_post:
thread["post_count"] += 1
thread["last_post_at"] = self.get_timestamp()
thread["last_post_by_uid"] = created_by_uid
await self.services.thread.save(thread)
# Update forum counters
forum = await self.services.forum.get(uid=forum_uid)
await forum.increment_post_count()
await self.services.forum.update_last_post(forum_uid, thread_uid)
await self.notify("post_created", {
"post": post,
"thread_uid": thread_uid,
"forum_uid": forum_uid
})
return post
raise Exception(f"Failed to create post: {post.errors}")
async def edit_post(self, post_uid, content, user_uid):
post = await self.get(uid=post_uid)
if not post:
return None
# Check permissions
user = await self.services.user.get(uid=user_uid)
if post["created_by_uid"] != user_uid and not user.get("is_admin"):
return None
post["content"] = content
post["edited_at"] = self.get_timestamp()
post["edited_by_uid"] = user_uid
if await self.save(post):
await self.notify("post_edited", post)
return post
return None
async def delete_post(self, post_uid, user_uid):
post = await self.get(uid=post_uid)
if not post:
return False
# Check permissions
user = await self.services.user.get(uid=user_uid)
if post["created_by_uid"] != user_uid and not user.get("is_admin"):
return False
# Don't allow deleting first post
if post["is_first_post"]:
return False
post["deleted_at"] = self.get_timestamp()
if await self.save(post):
await self.notify("post_deleted", post)
return True
return False
# services/post_like.py
class PostLikeService(BaseForumService):
mapper_name = "post_like"
async def toggle_like(self, post_uid, user_uid):
# Check if already liked
existing = await self.get(post_uid=post_uid, user_uid=user_uid)
if existing:
# Unlike
await self.delete(uid=existing["uid"])
# Update post like count
post = await self.services.post.get(uid=post_uid)
post["like_count"] = max(0, post["like_count"] - 1)
await self.services.post.save(post)
await self.notify("post_unliked", {
"post_uid": post_uid,
"user_uid": user_uid
})
return False
else:
# Like
like = await self.new()
like["post_uid"] = post_uid
like["user_uid"] = user_uid
if await self.save(like):
# Update post like count
post = await self.services.post.get(uid=post_uid)
post["like_count"] += 1
await self.services.post.save(post)
await self.notify("post_liked", {
"post_uid": post_uid,
"user_uid": user_uid
})
return True
return None
View Git Blame Copy Permalink