initial bot api

allows subscribing to new message events

Author: majestrate

sogs/base_config.py

@@ -60,3 +60,6 @@
 # If true, show recent messages for public rooms when accessed via a web browser.  If false only
 # show the QR code and URL but no recent messages.
 HTTP_SHOW_RECENT = True
+
+# sogs bot api listener addresses
+API_ADDRS = ['ipc://bot-api.sock']

sogs/events.py

@@ -0,0 +1,134 @@
+from collections import defaultdict
+
+from oxenmq import OxenMQ, AuthLevel
+
+from . import config
+from . import model
+
+import json
+
+# module level oxenmq
+_mq = None
+
+# pools for event propagation
+_pools = defaultdict(set)
+
+status_OK = 'OK'
+status_ERR = 'ERROR'
+
+# the events we are able to subscribe to
+EVENTS = ('message', 'joined', 'parted', 'banned', 'unbanned', 'deleted', 'uploaded')
+
+
+def event_name_valid(eventname):
+    """ return True if this event name is something well formed """
+    return eventname.lower() in EVENTS
+
+
+def _handle_subscribe_request(msg):
+    """ handle a request to subscribe to events as a bot"""
+    parts = msg.dataviews()
+    name = None
+    if len(parts) == 1:
+        name = parts[0].decode('ascii').lower()
+    if name is None:
+        raise Exception('no event name provided')
+    if event_name_valid(name):
+        raise Exception('invalid event name: {}'.format(name))
+    pool = _pools.get(name)
+    if msg.conn in pool:
+        raise Exception('already subscribed to {}'.format(name))
+    pool.append(msg.conn)
+
+
+def _handle_find_room(metric, query, encode=json.dumps):
+    """ find a room """
+    if metric == 'id':
+        room = model.Room(id=query)
+    elif metric == 'token':
+        room = model.Room(token=query)
+    else:
+        raise Exception("invalid query metric: {}".format(metric))
+    reply = dict()
+    reply['mods'] = room.get_mods()
+    reply['active_users'] = room.active_users()
+    for attr in model.Room.ALL_PROPS:
+        reply[attr] = getattr(room, attr)
+    return encode(reply)
+
+
+def _handle_find_user(metric, query, encode=json.dumps):
+    """ find a user """
+    if metric == 'id':
+        user = model.User(id=query)
+    elif metric == 'pubkey':
+        user = model.User(session_id=query)
+    else:
+        raise Exception("invalid query metric: {}".format(metric))
+    reply = dict()
+    for attr in model.User.ALL_PROPS:
+        reply[attr] = getattr(user, attr)
+    return encode(reply)
+
+
+def _handle_find_request(msg):
+    """ finds a user / room by id / token / pubkey """
+    parts = msg.dataviews()
+    if len(parts) != 3:
+        raise Exception('3 arguments required: entity-kind, query-metric, query-value')
+
+    decode = lambda x: x.decode('utf-8').lower()
+
+    kind = decode(parts[0])
+    metric = decode(parts[1])
+    query = decode(parts[2])
+
+    _kinds = {'room': _handle_find_room, 'user': _handle_find_user}
+
+    if kind in _kinds:
+        return _kinds[kind](metric, query)
+    raise Exception("cannot find a '{}' we dont have those".format(kind))
+
+
+def _propagate_event(eventname, *args):
+    """ propagate an event to everyone who cares about it """
+    assert event_name_valid(eventname)
+    global _mq
+    for conn in _pool.get(eventname):
+        _mq.command('sogs.event', eventname, *args)
+
+
+def _handle_request(handler, msg):
+    """ safely handle a request handler and catch any exceptions that happen and propagate them to the remote connection """
+    try:
+        retval = handler(msg)
+        if retval is None:
+            msg.reply(status_OK)
+        else:
+            msg.reply(status_OK, retval)
+    except Exception as ex:
+        msg.reply(status_ERROR, '{}'.format(ex))
+
+
+@postfork
+def start():
+    """ start event api """
+    global _mq
+    _mq = OxenMQ()
+    _bot_category = _mq.add_category('sogs', AuthLevel.none)
+    _bot_category.add_request_handler(
+        'sub', lambda msg: _handle_request(_handle_subscribe_request, msg)
+    )
+    _bot_category.add_request_handler(
+        'find', lambda msg: _handle_request(_handle_find_request, msg)
+    )
+
+    for addr in config.API_ADDRS:
+        # TODO: implement curve?
+        _mq.listen(addr, False)
+    _mq.start()
+
+
+# set up event notifiers
+for ev in EVENTS:
+    setattr(__module__, ev, lambda *args: _propagate_event(ev, *args))

sogs/model.py

@@ -3,6 +3,7 @@
 from . import config
 from . import db
 from . import utils
+from . import events
 
 import time
 
@@ -53,6 +54,20 @@ class Room:
         default_upload - True if default user permissions includes file upload permission
     """
 
+    """ set of all proprerties on a room """
+    ALL_PROPS = (
+        'id',
+        'token',
+        'name',
+        'description',
+        'created',
+        'updates',
+        'info_updates',
+        'default_read',
+        'default_write',
+        'default_upload',
+    )
+
     def __init__(self, row=None, *, id=None, token=None):
         """
         Constructs a room from a pre-retrieved row *or* via lookup of a room token or id.  When
@@ -320,6 +335,18 @@ class User:
         visible_mod - True if the user's admin/moderator status should be visible in rooms
     """
 
+    """ all accessable properties on a user """
+    ALL_PROPS = (
+        "id",
+        "session_id",
+        "created",
+        "last_active",
+        "banned",
+        "moderator",
+        "visible_mod",
+        "rooms",
+    )
+
     def __init__(self, row=None, *, id=None, session_id=None, autovivify=True, touch=False):
         """
         Constructs a user from a pre-retrieved row *or* a session id or user primary key value.
@@ -366,6 +393,16 @@ def __init__(self, row=None, *, id=None, session_id=None, autovivify=True, touch
         if touch:
             self._touch()
 
+    @property
+    def rooms(self):
+        result = list()
+        """ return the rooms this user is in by their id """
+        with db.conn as conn:
+            rows = conn.execute("SELECT room FROM room_users WHERE user = ?", (self.id,))
+            for row in rows:
+                result.append(row[0])
+        return result
+
     def _touch(self):
         db.conn.execute(
             """
@@ -530,6 +567,8 @@ def add_post_to_room(user_id, room_id, data, sig, rate_limit_size=5, rate_limit_
         result = conn.execute("SELECT posted, id FROM messages WHERE id = ?", [lastid])
         row = result.fetchone()
         msg = {'timestamp': utils.convert_time(row['posted']), 'server_id': row['id']}
+        # inform bot api of new post
+        events.message(user_id, room_id, utils.message_body(data))
         return msg