# retoor <retoor@molodetz.nl>
import asyncio
import json
import logging
import pathlib
from concurrent.futures import ProcessPoolExecutor
from snek.system.service import BaseService
from snek.system.template import sanitize_html
logger = logging.getLogger(__name__)
jinja2_env = None
template_path = pathlib.Path(__file__).parent.parent.joinpath("templates")
def render(context):
template = jinja2_env.get_template("message.html")
return sanitize_html(template.render(**context))
class ChannelMessageService(BaseService):
mapper_name = "channel_message"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._configured_indexes = False
self._executor_pools = {}
global jinja2_env
jinja2_env = self.app.jinja2_env
self._max_workers = 1
def get_or_create_executor(self, uid):
if not uid in self._executor_pools:
self._executor_pools[uid] = ProcessPoolExecutor(max_workers=self._max_workers)
logger.debug(f"Executors available: {len(self._executor_pools)}")
return self._executor_pools[uid]
def delete_executor(self, uid):
if uid in self._executor_pools:
self._executor_pools[uid].shutdown()
del self._executor_pools[uid]
async def maintenance(self):
while True:
changed = 0
async for message in self.find(is_final=False):
message["is_final"] = True
await self.save(message)
changed += 1
async for message in self.find(is_final=None):
message["is_final"] = False
await self.save(message)
changed += 1
if not changed:
break
async def create(self, channel_uid, user_uid, message, is_final=True):
logger.info(f"create: channel_uid={channel_uid}, user_uid={user_uid}, message_len={len(message) if message else 0}, is_final={is_final}")
model = await self.new()
model["channel_uid"] = channel_uid
model["user_uid"] = user_uid
model["message"] = message
model["is_final"] = is_final
context = {}
record = model.record
context.update(record)
user = await self.app.services.user.get(uid=user_uid)
if not user:
logger.error(f"create: user not found user_uid={user_uid}")
raise Exception("User not found")
context.update(
{
"user_uid": user["uid"],
"username": user["username"],
"user_nick": user["nick"],
"color": user["color"],
}
)
loop = asyncio.get_event_loop()
try:
context = json.loads(json.dumps(context, default=str))
logger.debug(f"create: rendering html for message uid={model['uid']}")
model["html"] = await loop.run_in_executor(self.get_or_create_executor(model["uid"]), render,context)
except Exception as ex:
logger.error(f"create: html rendering failed: {ex}")
logger.debug(f"create: saving message uid={model['uid']}")
if await super().save(model):
if not self._configured_indexes:
if not self.mapper.db["channel_message"].has_index(
["is_final", "user_uid", "channel_uid"]
):
self.mapper.db["channel_message"].create_index(
["is_final", "user_uid", "channel_uid"], unique=False
)
if not self.mapper.db["channel_message"].has_index(["uid"]):
self.mapper.db["channel_message"].create_index(["uid"], unique=True)
if not self.mapper.db["channel_message"].has_index(["deleted_at"]):
self.mapper.db["channel_message"].create_index(
["deleted_at"], unique=False
)
self._configured_indexes = True
if model['is_final']:
self.delete_executor(model['uid'])
logger.info(f"create: message created successfully uid={model['uid']}, channel={channel_uid}")
return model
logger.error(f"create: failed to save message channel={channel_uid}, errors={model.errors}")
raise Exception(f"Failed to create channel message: {model.errors}.")
async def to_extended_dict(self, message):
user = await self.services.user.get(uid=message["user_uid"])
if not user:
return {}
#if not message["html"].startswith("<chat-message"):
#message = await self.get(uid=message["uid"])
#await self.save(message)
return {
"uid": message["uid"],
"color": user["color"],
"user_uid": message["user_uid"],
"channel_uid": message["channel_uid"],
"user_nick": user["nick"],
"message": message["message"],
"created_at": message["created_at"],
"html": message["html"],
"username": user["username"],
}
async def save(self, model):
logger.debug(f"save: starting for uid={model['uid']}, is_final={model['is_final']}")
context = {}
context.update(model.record)
user = await self.app.services.user.get(model["user_uid"])
if not user:
logger.error(f"save: user not found user_uid={model['user_uid']}")
return False
context.update(
{
"user_uid": user["uid"],
"username": user["username"],
"user_nick": user["nick"],
"color": user["color"],
}
)
context = json.loads(json.dumps(context, default=str))
loop = asyncio.get_event_loop()
logger.debug(f"save: rendering html for uid={model['uid']}")
model["html"] = await loop.run_in_executor(self.get_or_create_executor(model["uid"]), render, context)
result = await super().save(model)
if result:
logger.debug(f"save: message saved successfully uid={model['uid']}")
else:
logger.warning(f"save: failed to save message uid={model['uid']}")
if model['is_final']:
self.delete_executor(model['uid'])
return result
async def offset(self, channel_uid, page=0, timestamp=None, page_size=30):
channel = await self.services.channel.get(uid=channel_uid)
if not channel:
return []
history_start_filter = ""
if channel["history_start"]:
history_start_filter = f" AND created_at > '{channel['history_start']}'"
results = []
offset = page * page_size
try:
if timestamp:
async for model in self.query(
f"SELECT * FROM channel_message WHERE channel_uid=:channel_uid AND created_at < :timestamp {history_start_filter} ORDER BY created_at DESC LIMIT :page_size OFFSET :offset",
{
"channel_uid": channel_uid,
"page_size": page_size,
"offset": offset,
"timestamp": timestamp,
},
):
results.append(model)
elif page > 0:
async for model in self.query(
f"SELECT * FROM channel_message WHERE channel_uid=:channel_uid WHERE created_at < :timestamp {history_start_filter} ORDER BY created_at DESC LIMIT :page_size",
{
"channel_uid": channel_uid,
"page_size": page_size,
"offset": offset,
"timestamp": timestamp,
},
):
results.append(model)
else:
async for model in self.query(
f"SELECT * FROM channel_message WHERE channel_uid=:channel_uid {history_start_filter} ORDER BY created_at DESC LIMIT :page_size OFFSET :offset",
{
"channel_uid": channel_uid,
"page_size": page_size,
"offset": offset,
},
):
results.append(model)
except Exception as ex:
logger.error(f"offset query failed: {ex}")
results.sort(key=lambda x: x["created_at"])
return results
View Git Blame Copy Permalink