SQLUsers.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from datetime import datetime, timedelta

import time, random, re, hashlib, base64
import logging

import smtplib
from email.mime.text import MIMEText

import _thread as thread

	
try:
	from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey, Boolean, Text, DateTime, UniqueConstraint, Index
	from sqlalchemy.orm import mapper, sessionmaker, relation, scoped_session
	from sqlalchemy.exc import IntegrityError
except ImportError as e:
	print("ERROR: sqlalchemy isn't installed: " + str(e))
	print("ERROR: please install sqlalchemy, on debian the command is sth. like: ")
	print("sudo apt-get install python-sqlalchemy")
	import sys
	sys.exit(1)


metadata = MetaData()

# Offline direct messages. The TTL matches channel_history's retention: the server holds
# private message content for as long as it holds public chat, and no longer. Past that
# the content is dropped and only a contentless tombstone survives, so a returning user
# still learns someone tried to reach them.
OFFLINE_MSG_TTL_DAYS = 14
OFFLINE_MSG_MAX_PER_PAIR = 50 # queued content messages one sender may hold for one recipient
##########################################
users_table = Table('users', metadata,
	Column('id', Integer, primary_key=True),
	Column('username', String(40), unique=True), # unicode
	Column('password', String(64)), # unicode(BASE64(ASCII)) (unicode is added by DB on write)
	Column('register_date', DateTime),
	Column('last_login', DateTime),
	Column('last_ip', String(15)), # would need update for ipv6
	Column('last_agent', String(254)),
	Column('last_sys_id', String(16)),
	Column('last_mac_id', String(16)),
	Column('ingame_time', Integer),
	Column('access', String(32)),
	Column('email', String(254), unique=True, nullable=True), # http://www.rfc-editor.org/errata_search.php?rfc=3696&eid=1690
	Column('bot', Integer),
	mysql_charset='utf8',
)

class User():
	def __init__(self, username, password, last_ip, email, access='agreement'):
		self.username = username
		self.password = password
		self.last_login = datetime.now()
		self.register_date = datetime.now()
		self.last_ip = last_ip
		self.last_agent = ""
		self.email = email
		self.ingame_time = 0
		self.bot = 0
		self.access = access # user, moderator, admin, bot, agreement, fresh
		self.last_sys_id = ""
		self.last_mac_id = ""

	def __repr__(self):
		return "<User('%s', '%s')>" % (self.username, self.password)

##########################################
verifications_table = Table('verifications', metadata,
	Column('id', Integer, primary_key=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('email', String(254), unique=True),
	Column('code', Integer),
	Column('expiry', DateTime),
	Column('attempts', Integer),
	Column('resends', Integer),
	Column('reason', Text),
	mysql_charset='utf8',
	)
class Verification(object):
	def __init__(self, user_id, email, digits, reason):
		self.user_id = user_id
		self.email = email
		assert(digits>=4)
		self.code = random.randint(10**(digits-1),10**(digits)-1)
		self.expiry = datetime.now() + timedelta(days=2)
		self.attempts = 0
		self.resends = 0
		self.reason = reason

	def __repr__(self):
		return "<Verification('%s', '%s', '%s', '%s', '%s', %i, %i)>" % (self.id, self.user_id, self.email, self.code, self.expiry, self.attempts, self.resends)
mapper(Verification, verifications_table)

##########################################
logins_table = Table('logins', metadata,
	Column('id', Integer, primary_key=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')), 
	Column('ip_address', String(15), nullable=False),
	Column('time', DateTime),
	Column('agent', String(64)),
	Column('last_sys_id', String(16)),
	Column('last_mac_id', String(16)),
	Column('local_ip', String(15)), # needs update for ipv6
	Column('country', String(4)),
	Column('end', DateTime),
	mysql_charset='utf8',
	)

class Login(object):
	def __init__(self, now, user_id, ip_address, agent, last_sys_id, last_mac_id, local_ip, country):
		self.user_id = user_id
		self.ip_address = ip_address
		self.time = now
		self.agent = agent
		self.last_sys_id = last_sys_id
		self.last_mac_id = last_mac_id
		self.local_ip = local_ip
		self.country = country

	def __repr__(self):
		return "<Login('%s', '%s')>" % (self.user_id, self.time)
mapper(Login, logins_table)
##########################################
bridged_users_table = Table('bridged_users', metadata,
	Column('id', Integer, primary_key=True),
	Column('external_id', String(20)),
	Column('location', String(20)),
	Column('external_username', String(20)),
	Column('last_bridged', DateTime),
	UniqueConstraint('external_id', 'location', name='uix_bridged_users_1'),
	UniqueConstraint('external_username', 'location', name='uix_bridged_users_2'),
	mysql_charset='utf8',
	)
class BridgedUser(object):
	def __init__(self, location, external_id, external_username, last_bridged):
		self.external_id = external_id
		self.location = location
		self.external_username = external_username
		self.last_bridged = last_bridged

	def __repr__(self):
		return "<BridgedUser('%s', '%s', '%s', '%s')>" % (self.id, self.external_id, self.location, self.last_bridged)
mapper(BridgedUser, bridged_users_table)
##########################################
renames_table = Table('renames', metadata,
	Column('id', Integer, primary_key=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('original', String(40)),
	Column('time', DateTime),
	mysql_charset='utf8',
	)
class Rename(object):
	def __init__(self, original):
		self.original = original
		self.time = datetime.now()

	def __repr__(self):
		return "<Rename of '%s'>" % (self.original)
mapper(Rename, renames_table)

##########################################
ignores_table = Table('ignores', metadata,
	Column('id', Integer, primary_key=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('ignored_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('reason', String(128)),
	Column('time', DateTime),
	mysql_charset='utf8',
	)
class Ignore(object):
	def __init__(self, user_id, ignored_user_id, reason):
		self.user_id = user_id
		self.ignored_user_id = ignored_user_id
		self.reason = reason
		self.time = datetime.now()

	def __repr__(self):
		return "<Ignore('%s', '%s', '%s', '%s')>" % (self.user_id, self.ignored_user_id, self.reason, self.time)
mapper(Ignore, ignores_table)

##########################################
friends_table = Table('friends', metadata,
	Column('id', Integer, primary_key=True),
	Column('first_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('second_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('time', DateTime),
	mysql_charset='utf8',
	)
class Friend(object):
	def __init__(self, first_user_id, second_user_id):
		self.first_user_id = first_user_id
		self.second_user_id = second_user_id
		self.time = datetime.now()

	def __repr__(self):
		return "<Friends('%s', '%s')>" % self.first_user_id, self.second_user_id
mapper(Friend, friends_table)
##########################################
friendRequests_table = Table('friendRequests', metadata,
	Column('id', Integer, primary_key=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('friend_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('msg', String(128)),
	Column('time', DateTime),
	mysql_charset='utf8',
	)
class FriendRequest(object):
	def __init__(self, user_id, friend_user_id, msg):
		self.user_id = user_id
		self.friend_user_id = friend_user_id
		self.msg = msg
		self.time = datetime.now()

	def __repr__(self):
		return "<FriendRequest('%s', '%s', '%s')>" % self.user_id, self.friend_user_id, self.msg
mapper(FriendRequest, friendRequests_table)
##########################################
mapper(User, users_table, properties={
	'logins':relation(Login, backref='user', cascade="all, delete, delete-orphan"),
	'renames':relation(Rename, backref='user', cascade="all, delete, delete-orphan"),
	## FIXME: all of these generate "Could not determine join condition between parent/child tables on relation User.XXXX"
	'ignores':relation(Ignore, cascade="all, delete, delete-orphan", foreign_keys=[Ignore.user_id]),
	'friends1':relation(Friend, cascade="all, delete, delete-orphan", foreign_keys=[Friend.first_user_id]),
	'friends2':relation(Friend, cascade="all, delete, delete-orphan", foreign_keys=[Friend.second_user_id]),
	'friend-requests-by-me':relation(FriendRequest, cascade="all, delete, delete-orphan", foreign_keys=[FriendRequest.user_id]),
	'friend-requests-for-me':relation(FriendRequest, cascade="all, delete, delete-orphan", foreign_keys=[FriendRequest.friend_user_id]),
	})

##########################################
channels_table = Table('channels', metadata,
	Column('id', Integer, primary_key=True),
	Column('name', String(40), unique=True),
	Column('key', String(32)),
	Column('owner_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True),
	Column('topic', Text),
	Column('topic_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True),
	Column('antispam', Boolean),
	Column('censor', Boolean),
	Column('store_history', Boolean),
	Column('last_used', DateTime),
	mysql_charset='utf8',
	)
class Channel(object):
	def __init__(self, name):
		self.name = name
		self.key = None
		self.owner_user_id = None
		self.topic = None
		self.topic_user_id = None
		self.antispam = False
		self.censor = False
		self.store_history = False
		self.last_used = None

	def __repr__(self):
		return "<Channel('%s')>" % self.name
mapper(Channel, channels_table)
##########################################
channelshistory_table = Table('channel_history', metadata,
	Column('id', Integer, primary_key=True),
	Column('channel_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('bridged_id', Integer, ForeignKey('bridged_users.id', onupdate='CASCADE', ondelete='CASCADE'), nullable=True),
	Column('time', DateTime),
	Column('msg', Text),
	Column('ex_msg', Boolean),
	mysql_charset='utf8',
	)
class ChannelHistory(object):
	def __init__(self, channel_id, user_id, bridged_id, time, msg, ex_msg):
		self.channel_id = channel_id
		self.user_id = user_id
		self.bridged_id = bridged_id
		self.time = time
		self.msg = msg
		self.ex_msg = ex_msg

	def __repr__(self):
		return "<ChannelHistory('%s')>" % self.channel_id
mapper(ChannelHistory, channelshistory_table)
##########################################
# Direct messages sent to a user who was offline, held until they next log in and then
# deleted. Content is transient by design: it is dropped after OFFLINE_MSG_TTL_DAYS, or
# when a sender exceeds OFFLINE_MSG_MAX_PER_PAIR queued messages for one recipient.
#
# A row with msg IS NULL is a TOMBSTONE: the content is gone but the recipient is still
# told that someone tried to reach them, so they know to ask rather than never finding
# out. There is at most one tombstone per (sender, recipient) pair - dropped_count says
# how many messages it stands for and time is the newest one's send time. Without that
# collapse, 50 expiring messages would become 50 tombstones.
offlinemessages_table = Table('offline_messages', metadata,
	Column('id', Integer, primary_key=True),
	Column('sender_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('recipient_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('time', DateTime),
	Column('msg', Text, nullable=True), # NULL == tombstone
	Column('ex_msg', Boolean),
	Column('dropped_count', Integer, default=0), # only meaningful on a tombstone
	Index('ix_offline_messages_recipient', 'recipient_user_id', 'id'), # login fetch
	Index('ix_offline_messages_pair', 'sender_user_id', 'recipient_user_id'), # per-pair cap count
	mysql_charset='utf8',
	)
class OfflineMessage(object):
	def __init__(self, sender_user_id, recipient_user_id, time, msg, ex_msg, dropped_count=0):
		self.sender_user_id = sender_user_id
		self.recipient_user_id = recipient_user_id
		self.time = time
		self.msg = msg
		self.ex_msg = ex_msg
		self.dropped_count = dropped_count

	def __repr__(self):
		return "<OfflineMessage(%s -> %s, %s)>" % (self.sender_user_id, self.recipient_user_id, "tombstone" if self.msg is None else "content")
mapper(OfflineMessage, offlinemessages_table)
##########################################
channelops_table = Table('channel_ops', metadata,
	Column('id', Integer, primary_key=True),
	Column('channel_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	)
class ChannelOp(object):
	def __init__(self, channel_id, user_id):
		self.channel_id = channel_id
		self.user_id = user_id

	def __repr__(self):
		return "<ChannelOp(%s,%s)>" % (self.channel_id, self.user_id)
mapper(ChannelOp, channelops_table)
##########################################
channelbans_table = Table('channel_bans', metadata,
	Column('id', Integer, primary_key=True),
	Column('channel_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('issuer_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('ip_address', String(15)),
	Column('expires', DateTime),
	Column('reason', Text)
	)
class ChannelBan(object):
	def __init__(self, channel_id, issuer_user_id, user_id, ip_address, expires, reason):
		self.channel_id = channel_id
		self.issuer_user_id = issuer_user_id
		self.user_id = user_id
		self.ip_address = ip_address
		self.expires = expires
		self.reason = reason

	def __repr__(self):
		return "<ChannelBan(%s,%s)>" % (self.channel_id, self.user_id)
mapper(ChannelBan, channelbans_table)
##########################################
channelbridgedbans_table = Table('channel_bridged_bans', metadata,
	Column('id', Integer, primary_key=True),
	Column('channel_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('issuer_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True),
	Column('bridged_id', Integer, ForeignKey('bridged_users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('expires', DateTime),
	Column('reason', Text)
	)
class ChannelBridgedBan(object):
	def __init__(self, channel_id, issuer_user_id, bridged_id, expires, reason):
		self.channel_id = channel_id
		self.issuer_user_id = issuer_user_id
		self.bridged_id = bridged_id
		self.expires = expires
		self.reason = reason

	def __repr__(self):
		return "<ChannelBridgedBan(%s,%s)>" % (self.channel_id, self.bridged_id)
mapper(ChannelBridgedBan, channelbridgedbans_table)
##########################################
channelmutes_table = Table('channel_mutes', metadata,
	Column('id', Integer, primary_key=True),
	Column('channel_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('issuer_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True),
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('expires', DateTime),
	Column('reason', Text)
	)
class ChannelMute(object):
	def __init__(self, channel_id, issuer_user_id, user_id, expires, reason):
		self.channel_id = channel_id
		self.issuer_user_id = issuer_user_id
		self.user_id = user_id
		self.expires = expires
		self.reason = reason

	def __repr__(self):
		return "<ChannelMute(%s,%s)>" % (self.channel_id, self.user_id)
mapper(ChannelMute, channelmutes_table)
##########################################
channelforwards_table = Table('channel_forwards', metadata,
	Column('id', Integer, primary_key=True),
	Column('channel_from_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	Column('channel_to_id', Integer, ForeignKey('channels.id', onupdate='CASCADE', ondelete='CASCADE')),
	UniqueConstraint('channel_from_id', 'channel_to_id', name='uix_channelforwards'),
	)
class ChannelForward(object):
	def __init__(self, channel_from_id, channel_to_id):
		self.channel_from_id = channel_from_id
		self.channel_to_id = channel_to_id

	def __repr__(self):
		return "<ChannelForward(%s,%s)>" % (self.channel_id_from, self.channel_id_to)
mapper(ChannelForward, channelforwards_table)
##########################################
ban_table = Table('ban', metadata, # server bans
	Column('id', Integer, primary_key=True),
	Column('issuer_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True), # user which set ban
	Column('user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE')), # user id which is banned (optional)
	Column('ip', String(60)), #ip which is banned (optional)
	Column('email', String(254)), #email which is banned (optional)
	Column('reason', Text),
	Column('end_date', DateTime),
	mysql_charset='utf8',
	)
class Ban(object):
	def __init__(self, issuer_user_id, duration, reason, user_id=None, ip=None, email=None):
		self.issuer_user_id = issuer_user_id
		self.user_id = user_id
		self.ip = ip
		self.email = email
		self.reason = reason
		self.end_date = datetime.now() + timedelta(duration)

	def __repr__(self):
		user_id_str = str(self.user_id)+', ' if self.user_id else ""
		ip_str = self.ip+', ' if self.ip else ""
		email_str = self.email+', ' if self.email else ""
		ban_str = user_id_str + ip_str + email_str
		return "<Ban: %s (%s, %s)>" % (ban_str, self.issuer_user_id, self.end_date)
mapper(Ban, ban_table)
##########################################
blacklisted_email_domain_table = Table('blacklisted_email_domains', metadata, # email domains that can't be used for account verification
	Column('id', Integer, primary_key=True),
	Column('issuer_user_id', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True), # user which set ban
	Column('domain', String(254), unique=True),
	Column('reason', Text),
	Column('start_time', DateTime),
	)
class BlacklistedEmailDomain(object):
	def __init__(self, issuer_user_id, domain, reason):
		self.issuer_user_id = issuer_user_id
		self.domain = domain
		self.reason = reason
		self.start_time = datetime.now()

	def __repr__(self):
		return "<Domain: %s (%s, since %s)>" % (self.domain, self.issuer_user_id, self.start_time)
mapper(BlacklistedEmailDomain, blacklisted_email_domain_table)
##########################################
min_spring_version_table = Table('min_spring_version', metadata,
	Column('id',Integer, primary_key=True),
	Column('min_spring_version', String(128)),
	Column('start_time', DateTime),
	)
class MinSpringVersion(object):
	def __init__(self, min_spring_version, start_time):
		self.min_spring_version = min_spring_version
		self.start_time = start_time
	
	def __repr__(self):
		return "<Version: %d (since %s)>" % (self.min_spring_version, self.start_time)
mapper(MinSpringVersion, min_spring_version_table)
##########################################

class session_manager():
	# on-demand, thread-local sessionmaker (Phase 3.1).
	# scoped_session gives each thread its own Session keyed by thread id, so DB work
	# run off the reactor thread via deferToThread cannot share a Session (which is not
	# thread-safe). On the single reactor thread this is behaviourally identical to the
	# previous single-shared-session model: lazily created, committed/closed per request
	# by the guards below (called from dataReceived / the LoopingCalls).
	def __init__(self, root, engine):
		self._root = root
		metadata.create_all(engine)
		self.sessionmaker = scoped_session(sessionmaker(bind=engine, autoflush=True))

	def sess(self):
		# returns this thread's session, creating it on first use
		return self.sessionmaker()

	# guarded access. registry.has() keeps the old "don't create a session just to
	# commit nothing" no-op semantics; these only ever touch the calling thread's session.
	def commit_guard(self):
		if self.sessionmaker.registry.has():
			self.sessionmaker.commit()
	def rollback_guard(self):
		if self.sessionmaker.registry.has():
			self.sessionmaker.rollback()
	def close_guard(self):
		if self.sessionmaker.registry.has():
			self.sessionmaker.remove()

##########################################
			
			
class OfflineClient():
	def __init__(self, sqluser):
		self.username = sqluser.username
		self.password = sqluser.password
		self.id = sqluser.id
		self.user_id = sqluser.id
		self.ingame_time = sqluser.ingame_time
		self.bot = sqluser.bot
		self.last_login = sqluser.last_login
		self.register_date = sqluser.register_date
		self.last_ip = sqluser.last_ip
		self.last_agent = sqluser.last_agent
		self.last_sys_id = sqluser.last_sys_id
		self.last_mac_id = sqluser.last_mac_id
		self.access = sqluser.access
		self.email = sqluser.email

class UsersHandler:
	def __init__(self, root):
		self._root = root

		# 1.2: bounded TTL cache of offline-user lookups. Stores positive hits only
		# (OfflineClient snapshots) keyed separately by username and by id. This only
		# ever serves *offline* lookups: callers go through DataHandler which checks the
		# online dicts first, and login/auth (check_login_user, login_user) query User
		# directly and never touch this cache, so a stale entry can't bypass auth.
		# Invalidated precisely at every User-row write below; bulk writes flush it.
		self._user_cache_ttl = 120  # seconds
		self._user_cache_max = 500  # entries per dict
		self._user_cache_by_name = {}  # username -> (OfflineClient, expiry)
		self._user_cache_by_id = {}    # user_id  -> (OfflineClient, expiry)

	def sess(self):
		return self._root.session_manager.sess()

	def _user_cache_get(self, store, key):
		hit = store.get(key)
		if hit is None:
			return None
		client, expiry = hit
		if expiry < time.time():
			del store[key]
			return None
		return client

	def _user_cache_put(self, client):
		# cache a positive hit under both keys; FIFO-evict when a dict is full
		expiry = time.time() + self._user_cache_ttl
		for store, key in ((self._user_cache_by_name, client.username), (self._user_cache_by_id, client.id)):
			if key not in store and len(store) >= self._user_cache_max:
				store.pop(next(iter(store)))
			store[key] = (client, expiry)

	def invalidate_user_cache(self, user_id=None, username=None):
		if username is not None:
			self._user_cache_by_name.pop(username, None)
		if user_id is not None:
			self._user_cache_by_id.pop(user_id, None)

	def flush_user_cache(self):
		self._user_cache_by_name.clear()
		self._user_cache_by_id.clear()

	def clientFromID(self, user_id):
		cached = self._user_cache_get(self._user_cache_by_id, user_id)
		if cached is not None:
			return cached
		entry = self.sess().query(User).filter(User.id==user_id).first()
		if not entry: return None
		client = OfflineClient(entry)
		self._user_cache_put(client)
		return client

	def clientFromUsername(self, username):
		cached = self._user_cache_get(self._user_cache_by_name, username)
		if cached is not None:
			return cached
		entry = self.sess().query(User).filter(User.username==username).first()
		if not entry: return None
		client = OfflineClient(entry)
		self._user_cache_put(client)
		return client

	def remaining_ban_str(self, dbban, now):
		timeleft = int((dbban.end_date - now).total_seconds())
		remaining = 'less than one hour remaining'
		if timeleft > 60*60*24*900:
			remaining = ''
		elif timeleft > 60*60*24:
			remaining = '%s days remaining' % (int(timeleft / (60 * 60 * 24)))
		elif timeleft > 60*60:
			remaining = '%s hours remaining' % (int(timeleft / (60 * 60)))
		return remaining

	def check_banned(self, username, ip):
		# returns (banned, reason, dbuser); dbuser is the live User row (or None) so the
		# caller can hand it straight to login_user without re-querying it (1.4).
		dbuser = self.sess().query(User).filter(User.username == username).first()
		if not dbuser:
			return False, "", None
		now = datetime.now()
		dbban = self._root.bandb.check_ban(dbuser.id, ip, dbuser.email, now)
		if dbban and not dbuser.access=='admin':
			reason = 'You are banned: (%s), ' %(dbban.reason)
			reason += self.remaining_ban_str(dbban, now)
			return True, reason, dbuser
		return False, "", dbuser
		
	def check_login_user(self, username, password):
		# password here is unicode(BASE64(MD5(...))), matches the register_user DB encoding
		dbuser = self.sess().query(User).filter(User.username == username).first()
		if (not dbuser):
			return False, 'Invalid username or password'
		if dbuser.username != username:
			# user tried to login with wrong upper/lower case somewhere in their username
			return False, "Invalid username -- did you mean '%s'" % dbuser.username
		if dbuser.password != password:
			return False, 'Invalid username or password'
		return True, ""
		
	def login_user(self, dbuser, ip, agent, last_sys_id, last_mac_id, local_ip, country):
		# dbuser is the live User row already loaded by check_banned (1.4), so we don't
		# re-query it here. All field updates plus the Login record are flushed in a
		# single commit/transaction (1.3).
		now = datetime.now()
		dbuser.logins.append(Login(now, dbuser.id, ip, agent, last_sys_id, last_mac_id, local_ip, country))
		dbuser.last_ip = ip
		dbuser.last_agent = agent
		dbuser.last_sys_id = last_sys_id
		dbuser.last_mac_id = last_mac_id
		dbuser.last_login = now

		# capture identifiers before commit; expire_on_commit would otherwise force a
		# reload query just to read them back
		uid, uname = dbuser.id, dbuser.username
		self.sess().commit()
		self.invalidate_user_cache(uid, uname)
		return dbuser

	# --- 3.1 async-login workers -------------------------------------------------
	# precheck_login and do_login run on a deferToThread worker thread. They do PURE
	# DB I/O and return plain data (tuples / an OfflineClient snapshot) so the reactor
	# callback never touches an ORM row bound to the worker's session. They must NOT
	# mutate shared in-memory state (incl. the user cache) - the reactor callback does
	# that.

	def precheck_login(self, username, password, ip):
		# auth + ban check, mirroring the login_now order: check_login_user first,
		# and only check the ban if the credentials are valid (matching the original
		# control flow, where a failed credential check returns before the ban query).
		# Returns plain values; the live User row from check_banned stays in this thread.
		login_ok, login_reason = self.check_login_user(username, password)
		banned, ban_reason = False, ""
		if login_ok:
			banned, ban_reason, _dbuser = self.check_banned(username, ip)
		return login_ok, login_reason, banned, ban_reason

	def do_login(self, username, ip, agent, last_sys_id, last_mac_id, local_ip, country):
		# write the login: re-load the user (the precheck row belonged to another
		# thread/session and cannot cross threads), update last_* fields, append the
		# Login record, commit, and return a (snapshot, ignored_user_ids) pair. The
		# ignore list is fetched here, on this worker thread, so the reactor callback
		# (_SendLoginInfo) does not block the event loop on a synchronous
		# get_ignored_user_ids query. Returns (None, []) if the user vanished
		# (rename/delete) between precheck and here. Cache invalidation is left to the
		# reactor callback (shared-state mutation).
		dbuser = self.sess().query(User).filter(User.username == username).first()
		if not dbuser:
			return None, []
		now = datetime.now()
		dbuser.logins.append(Login(now, dbuser.id, ip, agent, last_sys_id, last_mac_id, local_ip, country))
		dbuser.last_ip = ip
		dbuser.last_agent = agent
		dbuser.last_sys_id = last_sys_id
		dbuser.last_mac_id = last_mac_id
		dbuser.last_login = now
		# snapshot while the row is attached and current, before commit expires it
		snapshot = OfflineClient(dbuser)
		ignored = self.get_ignored_user_ids(dbuser.id)
		self.sess().commit()
		return snapshot, ignored

	def do_change_password(self, username, cur_password, new_password):
		# 3.1 worker: re-verify the current password and write the new one as ONE
		# uncommitted transaction (atomic check-then-set; _run_db owns the commit).
		# Returns ('denied', reason) without writing on a credential mismatch (or a
		# vanished user), else ('ok', uid). Cache-free - the reactor callback does the
		# 1.2 user-cache invalidation; this must not touch shared in-memory state.
		good, reason = self.check_login_user(username, cur_password)
		if not good:
			return ('denied', reason)
		dbuser = self.sess().query(User).filter(User.username==username).first()
		dbuser.password = new_password
		return ('ok', dbuser.id)

	def set_bot(self, user_id, is_bot):
		dbuser = self.sess().query(User).filter(User.id==user_id).first()
		if dbuser:
			dbuser.bot = 1 if is_bot else 0
			uname = dbuser.username
			self.sess().commit()
			self.invalidate_user_cache(user_id, uname)


	def do_end_session(self, user_id):
		# 3.1 worker: PURE DB, one uncommitted transaction (defer_db/_run_db owns the
		# commit and retries on serialization errors). Returns the plain username for the
		# reactor callback to invalidate the 1.2 cache with; never touches the cache itself.
		# Idempotent: if the most recent login is already ended, do nothing and return None.
		entry = self.sess().query(User).filter(User.id==user_id).first()
		if not entry or not entry.logins or entry.logins[-1].end:
			return None
		entry.logins[-1].end = datetime.now()
		entry.last_login = datetime.now() # in real its last online / last seen
		return entry.username

	def check_user_name(self, user_name):
		if len(user_name) > 20: return False, 'Username too long'
		if self._root.censor:
			if not self._root.SayHooks._nasty_word_censor(user_name):
				return False, 'Name failed to pass profanity filter.'
		return True, ""

	def check_register_user(self, username, email=None, ip_address=None):
		assert(type(username) == str)

		status, reason = self.check_user_name(username)
		if not status:
			return False, reason
		dbuser = self.sess().query(User).filter(User.username == username).first()
		if dbuser:
			return False, 'Username is already in use.'
		if email:
			dbemail = self.sess().query(User).filter(User.email == email).first()
			if dbemail:
				return False, 'Email address is already in use.'
		if ip_address:
			ipban = self._root.bandb.check_ban(None, ip_address)
			if ipban:
				return False, 'Account registration failed: %s' % ipban.reason
		return True, ""

	def register_user(self, username, password, ip, email):
		# note: password here is BASE64(MD5(...)) and already in unicode
		# assume check_register_user was already called
		entry = User(username, password, ip, email)

		self.sess().add(entry)
		self.sess().commit()
		# positives-only cache: a brand-new id has no stale entry, so invalidating the
		# username alone covers the recreated-name edge without forcing a reload for id
		self.invalidate_user_cache(username=username)
		return True, 'Account registered successfully.'

	def do_register_insert(self, username, password, ip, email):
		# 3.1 worker: INSERT the new account as one uncommitted transaction (defer_db/_run_db
		# owns the commit). check_register_user already ran on the reactor, but a concurrent
		# REGISTER or RENAMEACCOUNT can still win the same username/email in the gap before
		# this insert, so catch the unique-constraint violation and report it as ('taken',)
		# rather than letting a 1062 surface as a server error. On success returns the new id
		# for the reactor callback; never touches the 1.2 cache (that stays on the reactor).
		entry = User(username, password, ip, email)
		self.sess().add(entry)
		try:
			self.sess().flush()  # forces the INSERT now so we can catch the dup here
		except IntegrityError:
			self.sess().rollback()
			return ('taken',)
		return ('ok', entry.id)

	def do_rename_account(self, username, newname):
		# 3.1 worker: PURE DB, ONE uncommitted transaction (_run_db owns the commit + retry).
		# Replaces the old self-committing rename_user; returns verdict tuples and leaves cache
		# invalidation to the reactor callback. The early uniqueness query is load-bearing for
		# the case-insensitive
		# collation behaviour: it matches the caller's OWN row on a case-only self-rename, so
		# that is denied exactly as today (relying solely on the UNIQUE constraint would let a
		# self-rename to a case variant succeed - a row's UPDATE never conflicts with itself).
		# The IntegrityError catch handles the residual race where two concurrent renames both
		# pass the early query: the UNIQUE(username) violation is 1062, which is NOT in _run_db's
		# retry set, so we catch it here, roll back, and deny rather than retry a doomed write.
		# The newname==username exact match is handled on the reactor before deferring.
		results = self.sess().query(User).filter(User.username==newname).first()
		if results:
			return ('denied', 'Username already exists.')
		entry = self.sess().query(User).filter(User.username==username).first()
		if not entry:
			return ('denied', 'You don\'t seem to exist anymore. Contact an admin or moderator.')
		try:
			entry.renames.append(Rename(username))
			entry.username = newname
			uid = entry.id
			self.sess().flush() # force the UNIQUE check to fire HERE so 1062 is catchable
		except IntegrityError:
			self.sess().rollback()
			return ('denied', 'Username already exists.')
		return ('ok', uid, newname)

	def save_user(self, obj):
		# assert(isinstance(obj, User) or isinstance(obj, Client))
		entry = self.sess().query(User).filter(User.username==obj.username).first()
		if (entry != None):
			# caller might have changed these!
			entry.password = obj.password
			entry.ingame_time = obj.ingame_time
			entry.access = obj.access
			entry.bot = obj.bot
			entry.last_sys_id = obj.last_sys_id
			entry.last_mac_id = obj.last_mac_id
			entry.email = obj.email

		uid = entry.id if entry is not None else None
		self.sess().commit()
		self.invalidate_user_cache(uid, obj.username)

	def do_set_ingame_time(self, username, ingame_time):
		# 3.1 worker: PURE DB, one uncommitted transaction. MYSTATUS only ever changes
		# ingame_time; we write the ABSOLUTE total (the reactor already incremented the
		# in-memory client.ingame_time) so a _run_db retry re-applies the same value rather
		# than double-counting. Other User fields are left untouched on purpose - this avoids
		# clobbering a field a concurrent handler (e.g. CHANGEPASSWORD) may have just written,
		# the race that deferring the old all-field save_user would have opened. Returns the
		# user id for the reactor callback to invalidate the 1.2 cache with.
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry is None:
			return None
		entry.ingame_time = ingame_time
		return entry.id

	def do_change_email(self, username, newmail):
		# 3.1 worker: PURE DB, one uncommitted transaction (_run_db owns the commit + retry).
		# CHANGEEMAIL only changes email, so we write ONLY that field rather than mirroring the
		# all-field save_user: writing ingame_time/password/access here would clobber a value a
		# concurrent handler (MYSTATUS's do_set_ingame_time, CHANGEPASSWORD) may have just written
		# - the same clobber do_set_ingame_time avoids. users.email is UNIQUE, so the residual
		# change-to-taken race that slips past the reactor pre-check surfaces as a 1062, which is
		# NOT in _run_db's retry set; flush() forces it to fire HERE so we catch it, roll back,
		# and deny rather than retry a doomed write. The verification gate stays on the reactor
		# (it self-commits, so it can't live inside this single-transaction unit). Returns the
		# user id for the reactor callback to invalidate the 1.2 cache with.
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry is None:
			return ('denied', 'You don\'t seem to exist anymore. Contact an admin or moderator.')
		try:
			entry.email = newmail
			uid = entry.id
			self.sess().flush() # force the UNIQUE check to fire HERE so 1062 is catchable
		except IntegrityError:
			self.sess().rollback()
			return ('denied', "another user is already registered to the email address '%s'" % newmail)
		return ('ok', uid, newmail)

	def do_confirm_agreement(self, username):
		# 3.1 worker: PURE DB, one uncommitted transaction (_run_db owns the commit + retry).
		# CONFIRMAGREEMENT only flips access 'agreement'->'user', so we write ONLY that field
		# rather than mirroring the all-field save_user (which would clobber a value a concurrent
		# handler may have just written, the same clobber do_set_ingame_time avoids). access is
		# not unique, so there is no 1062 path here. Returns the user id for the reactor callback
		# to invalidate the 1.2 cache with, or None if the account vanished.
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry is None:
			return None
		entry.access = 'user'
		return entry.id

	def do_scrub_account(self, user_id, hashed_password):
		# 3.1 worker: PURE DB, one uncommitted transaction (_run_db owns the commit + retry).
		# DELETEACCOUNT's account scrub, folding the old save_user + reset_password writes into a
		# single atomic users-row write: zero ingame_time/bot, force access 'user', lock the
		# account with a fresh password (generated on the reactor and passed in hashed so a
		# _run_db retry re-applies the SAME value - never regenerated, the MYSTATUS idempotency
		# precedent), and clear the email so the account cannot be recovered. email is set to
		# NULL, not '': users.email is UNIQUE, so '' on a second deleted account would collide
		# with a first that still holds '' (1062); NULL is exempt from UNIQUE and equally blocks
		# recovery, matching the REGISTER email=None precedent. flush() forces any residual
		# integrity error to fire HERE. Keyed by user_id because the caller may hold an offline
		# snapshot. Returns ('ok', uid), or ('gone',) if the account vanished.
		entry = self.sess().query(User).filter(User.id==user_id).first()
		if entry is None:
			return ('gone',)
		entry.ingame_time = 0
		entry.bot = 0
		entry.access = 'user'
		entry.email = None
		entry.password = hashed_password
		self.sess().flush()
		return ('ok', entry.id)

	def do_set_password(self, user_id, hashed_password, new_email=None):
		# 3.1 worker: PURE DB, one uncommitted transaction (_run_db owns the commit + retry).
		# Shared by RESETPASSWORD and RESETUSERPASSWORD: writes a fresh password (generated on
		# the reactor and passed in hashed so a _run_db retry re-applies the SAME value rather
		# than regenerating; the RAW password and the SMTP send stay in the reactor callback, so
		# a failed or retried write never emails). RESETUSERPASSWORD's add-email path also passes
		# new_email; users.email is UNIQUE, so flush() forces a 1062 to fire HERE where we catch
		# it, roll back and deny rather than letting _run_db retry a doomed write (1062 is not in
		# its retry set). Keyed by user_id (the caller may hold an offline snapshot). Returns
		# ('ok', username, email) on success, or ('denied', reason) if the account vanished or
		# the new email is already taken.
		entry = self.sess().query(User).filter(User.id==user_id).first()
		if entry is None:
			return ('denied', "User no longer exists")
		try:
			entry.password = hashed_password
			if new_email is not None:
				entry.email = new_email
			uname = entry.username
			email = entry.email
			self.sess().flush()
		except IntegrityError:
			self.sess().rollback()
			return ('denied', "another user is already registered to the email address '%s'" % new_email)
		return ('ok', uname, email)

	def generate_password(self):
		# reactor-side helper (NO DB): mint a fresh random password ONCE and return (raw, hashed).
		# The hashed value goes to the do_set_password / do_scrub_account worker so a _run_db retry
		# re-applies the SAME value rather than regenerating it; the raw value is emailed from the
		# reactor callback after the write commits. Mirrors VerificationsHandler.reset_password's
		# generation so the password format is unchanged.
		char_set = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890!£$%^&*?"
		raw = ""
		for i in range(0, 10):
			raw += random.choice(char_set)
		hash = hashlib.md5()
		hash.update(str.encode(raw))
		hashed = base64.b64encode(hash.digest()).decode()
		assert(self._root.protocol._validPasswordSyntax(hashed))
		return raw, hashed

	def get_user_id_with_email(self, email):
		if email == '':
			return False, 'Email address is blank'
		response = self.sess().query(User).filter(User.email == email)
		dbuser = response.first()
		if not dbuser:
			return False, 'No user with email address %s was found' % email
		return True, dbuser.id

	def confirm_agreement(self, client):
		entry = self.sess().query(User).filter(User.username==client.username).first()
		if entry: entry.access = 'user'
		uid = entry.id if entry else None
		self.sess().commit()
		self.invalidate_user_cache(uid, client.username)

	def get_lastlogin(self, username):
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry: return True, entry.last_login
		else: return False, 'User not found.'

	def get_registration_date(self, username):
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry and entry.register_date: return True, entry.register_date
		else: return False, 'User or date not found in database'

	def get_ingame_time(self, username):
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry: return True, entry.ingame_time
		else: return False, 'User not found in database'

	def get_account_access(self, username):
		entry = self.sess().query(User).filter(User.username==username).first()
		if entry:
			return True, entry.access
		else: return False, 'User not found in database'

	def find_ip(self, ip):
		results = self.sess().query(User).filter(User.last_ip==ip)
		return results

	def do_find_ip(self, ip):
		# 3.1 worker: PURE DB. find_ip() returns live, session-bound User rows (a Query);
		# materialize them to plain (username, last_login-isoformat-or-None) tuples HERE so the
		# reactor callback never touches a row bound to this worker's session. The online/offline
		# distinction (username in usernames) and the SERVERMSG formatting stay on the reactor.
		return [(u.username, u.last_login.isoformat() if u.last_login else None) for u in self.find_ip(ip)]

	def get_ip(self, username):
		entry = self.sess().query(User).filter(User.username==username).first()
		if not entry:
			return None
		return entry.last_ip

	def list_mods(self):
		response_mods = self.sess().query(User).filter('mod' == User.access)
		mods = ""
		for user in response_mods:
			mods += user.username + " "
		admins = ""
		response_admins = self.sess().query(User).filter('admin' == User.access)
		for user in response_admins:
			admins += user.username + " "
		return admins, mods

	def remove_user(self, user):
		entry = self.sess().query(User).filter(User.username==user).first()
		if not entry:
			return False, 'User not found.'
		user_id = entry.id
		self.sess().delete(entry)
		self.sess().commit()
		self.invalidate_user_cache(user_id, user)
		return True, 'Success.'

	def clean(self):
		now = datetime.now()
		#delete users:
		# which didn't accept agreement after three days
		response = self.sess().query(User).filter(User.register_date < now - timedelta(days=3)).filter(User.access == "agreement")
		logging.info("deleting %i users who failed to verify registration", response.count())
		response.delete(synchronize_session=False)

		# which have no ingame time, last login > 1 month ago, not bot, not mod
		response = self.sess().query(User).filter(User.ingame_time == 0).filter(User.last_login < now - timedelta(days=28)).filter(User.bot == 0).filter(User.access == "user")
		logging.info("deleting %i inactive users with no ingame time", response.count())
		response.delete(synchronize_session=False)

		# last login > 5 years
		response = self.sess().query(User).filter(User.last_login < now - timedelta(days=1825))
		logging.info("deleting %i very inactive users", response.count())
		response.delete(synchronize_session=False)

		# old messages > 2 weeks
		response = self.sess().query(ChannelHistory).filter(ChannelHistory.time < now - timedelta(days=14))
		logging.info("deleting %i channel history messages", response.count())
		response.delete(synchronize_session=False)

		self.expire_offline_messages(now)

		self.sess().commit()
		self.flush_user_cache()

	def expire_offline_messages(self, now=None):
		# Drop offline message CONTENT past the TTL, but leave one tombstone per
		# (sender, recipient) pair behind so a user returning after a long absence still
		# learns someone tried to reach them. Tombstones themselves never expire here -
		# they live until delivered, which is the whole point of them.
		#
		# Callers: clean(), which commits. Kept commit-free so it stays one unit with it.
		if now is None:
			now = datetime.now()
		cutoff = now - timedelta(days=OFFLINE_MSG_TTL_DAYS)
		expired = self.sess().query(OfflineMessage).filter(OfflineMessage.time < cutoff).filter(OfflineMessage.msg != None).all()
		if not expired:
			return 0
		logging.info("expiring %i offline messages into tombstones", len(expired))
		# Collapse per pair first: N expiring messages from one sender must become ONE
		# tombstone counting N, not N tombstones.
		per_pair = {}
		for m in expired:
			key = (m.sender_user_id, m.recipient_user_id)
			count, newest = per_pair.get(key, (0, m.time))
			per_pair[key] = (count + 1, max(newest, m.time))
			self.sess().delete(m)
		self.sess().flush() # the deletes must land before _bump_offline_tombstone re-reads
		for (sender_id, recipient_id), (count, newest) in per_pair.items():
			self._bump_offline_tombstone(sender_id, recipient_id, newest, count)
		return len(expired)

	def audit_access(self):
		now = datetime.now()
		# remove botflags from clients that didn't log in for 1 year
		response = self.sess().query(User).filter(User.last_login < now - timedelta(days=365)).filter(User.bot == 1)
		logging.info("removing %i botflags from inactive hosts", response.count())
		for user in response:
			user.bot = 0
		
		# remove moderator/admin access from clients that didn't log in for 1 year:
		response = self.sess().query(User).filter(User.last_login < now - timedelta(days=365)).filter(User.access == "admin")
		logging.info("removing %i inactive admins", response.count())
		for user in response:
			user.access = "user"
		response = self.sess().query(User).filter(User.last_login < now - timedelta(days=365)).filter(User.access == "mod")
		logging.info("removing %i inactive mods", response.count())
		for user in response:
			user.access = "user"

		self.sess().commit()
		self.flush_user_cache()

	def ignore_user(self, user_id, ignore_user_id, reason=None):
		entry = Ignore(user_id, ignore_user_id, reason)
		self.sess().add(entry)
		self.sess().commit()

	def unignore_user(self, user_id, unignore_user_id):
		entry = self.sess().query(Ignore).filter(Ignore.user_id == user_id).filter(Ignore.ignored_user_id == unignore_user_id).one()
		self.sess().delete(entry)
		self.sess().commit()

	# returns id-s of users who had their ignore removed
	def globally_unignore_user(self, unignore_user_id):
		q = self.sess().query(Ignore).filter(Ignore.ignored_user_id == unignore_user_id)
		userids = [ignore.user_id for ignore in q.all()]
		# could be done in one query + hook, fix if bored
		self.sess().query(Ignore).filter(Ignore.ignored_user_id == unignore_user_id).delete()
		self.sess().commit()
		return userids

	def is_ignored(self, user_id, ignore_user_id):
		exists = self.sess().query(Ignore).filter(Ignore.user_id == user_id).filter(Ignore.ignored_user_id == ignore_user_id).count() > 0
		return exists

	def get_ignore_list(self, user_id):
		users = self.sess().query(Ignore).filter(Ignore.user_id == user_id).all()
		users = [(user.ignored_user_id, user.reason) for user in users]
		return users

	def get_ignored_user_ids(self, user_id):
		user_ids = self.sess().query(Ignore.ignored_user_id).filter(Ignore.user_id == user_id).all()
		user_ids = [user_id for user_id, in user_ids]
		return user_ids

	def friend_users(self, user_id, friend_user_id):
		entry = Friend(user_id, friend_user_id)
		self.sess().add(entry)
		self.sess().commit()

	def unfriend_users(self, first_user_id, second_user_id):
		self.sess().query(Friend).filter(Friend.first_user_id == first_user_id).filter(Friend.second_user_id == second_user_id).delete()
		self.sess().query(Friend).filter(Friend.second_user_id == first_user_id).filter(Friend.first_user_id == second_user_id).delete()
		self.sess().commit()

	def are_friends(self, first_user_id, second_user_id):
		q1 = self.sess().query(Friend).filter(Friend.first_user_id == first_user_id)
		q2 = self.sess().query(Friend).filter(Friend.second_user_id == second_user_id)
		exists = q1.union(q2).count() > 0
		return exists

	def get_friend_user_ids(self, user_id):
		q1 = self.sess().query(Friend.second_user_id).filter(Friend.first_user_id == user_id)
		q2 = self.sess().query(Friend.first_user_id).filter(Friend.second_user_id == user_id)
		user_ids = q1.union(q2).all()
		user_ids = [user_id for user_id, in user_ids]
		return user_ids

	def has_friend_request(self, user_id, friend_user_id):
		request = self.sess().query(FriendRequest).filter(FriendRequest.user_id == user_id).filter(FriendRequest.friend_user_id == friend_user_id)
		exists = request.count() > 0
		return exists

	def add_friend_request(self, user_id, friend_user_id, msg=None):
		entry = FriendRequest(user_id, friend_user_id, msg)
		self.sess().add(entry)
		self.sess().commit()

	def remove_friend_request(self, user_id, friend_user_id):
		self.sess().query(FriendRequest).filter(FriendRequest.user_id == user_id).filter(FriendRequest.friend_user_id == friend_user_id).delete()
		self.sess().commit()

	# this returns all friend requests sent _to_ user_id
	def get_friend_request_list(self, user_id):
		reqs = self.sess().query(FriendRequest).filter(FriendRequest.friend_user_id == user_id).all()
		users = [(req.user_id, req.msg) for req in reqs]
		return users

	# 3.1 worker units for the friend mutations. Each runs on a thread-pool worker and does
	# the whole check-then-act as ONE uncommitted transaction: DataHandler._run_db commits it
	# once and can safely retry the entire unit on a transient serialization error. They must
	# NOT call the self-committing helpers above (friend_users / add_friend_request / ...),
	# because an intermediate commit would be re-applied on retry. They take/return plain data
	# only (ids, tuples) - never a live ORM row or an in-memory client - so they are
	# thread-safe; the reactor-thread callback does all client notification.
	def _user_id_from_username(self, username):
		# cache-free, case-sensitive username -> id resolution. The offline-user cache in
		# clientFromUsername must not be touched off the reactor thread, so query directly.
		# db collation is case-insensitive, so require an exact match (mirrors clientFromUsername).
		row = self.sess().query(User.id, User.username).filter(User.username == username).first()
		if not row or row[1] != username:
			return None
		return row[0]

	def do_friend_request(self, from_id, target_username, msg):
		target_id = self._user_id_from_username(target_username)
		if target_id is None:
			return ('nouser',)
		if self.are_friends(from_id, target_id):
			return ('already_friends',)
		if self.is_ignored(target_id, from_id):
			# target ignores the sender; the ignore-set is kept in sync with this table for
			# online users too, so the db check is correct whether the target is on- or offline
			return ('silent',)
		if self.has_friend_request(from_id, target_id):
			return ('silent',)
		self.sess().add(FriendRequest(from_id, target_id, msg))
		return ('ok', target_id)

	def do_accept_friend_request(self, accepter_id, target_username):
		target_id = self._user_id_from_username(target_username)
		if target_id is None:
			return ('no_request',)
		if not self.has_friend_request(target_id, accepter_id):
			return ('no_request',)
		self.sess().add(Friend(accepter_id, target_id))
		self.sess().query(FriendRequest).filter(FriendRequest.user_id == target_id).filter(FriendRequest.friend_user_id == accepter_id).delete()
		return ('ok', target_id)

	def do_decline_friend_request(self, decliner_id, target_username):
		target_id = self._user_id_from_username(target_username)
		if target_id is None:
			return ('no_request',)
		if not self.has_friend_request(target_id, decliner_id):
			return ('no_request',)
		self.sess().query(FriendRequest).filter(FriendRequest.user_id == target_id).filter(FriendRequest.friend_user_id == decliner_id).delete()
		return ('ok',)

	def do_unfriend(self, user_id, target_username):
		target_id = self._user_id_from_username(target_username)
		if target_id is None:
			return ('nouser',)
		self.sess().query(Friend).filter(Friend.first_user_id == user_id).filter(Friend.second_user_id == target_id).delete()
		self.sess().query(Friend).filter(Friend.second_user_id == user_id).filter(Friend.first_user_id == target_id).delete()
		return ('ok', target_id)

	def _user_access_from_username(self, username):
		# cache-free resolve returning (id, access) - the IGNORE mod/admin check needs the
		# access string, which _user_id_from_username does not return. Mirrors that helper:
		# case-sensitive exact match (db collation is case-insensitive, so re-check in python).
		row = self.sess().query(User.id, User.username, User.access).filter(User.username == username).first()
		if not row or row[1] != username:
			return None
		return (row[0], row[2])

	def do_ignore_insert(self, user_id, ignore_user_id, reason):
		# bare INSERT as ONE uncommitted unit (the reactor already validated). There is NO
		# unique constraint on (user_id, ignored_user_id), so a concurrent same-target IGNORE
		# from a second connection can still create a duplicate row; do_unignore_delete below
		# tolerates and clears duplicates. A future migration should add the unique constraint.
		self.sess().add(Ignore(user_id, ignore_user_id, reason))

	def do_unignore_delete(self, user_id, unignore_user_id):
		# tolerant DELETE (no .one(), unlike unignore_user) so a duplicate row from the
		# no-unique-constraint race cannot raise MultipleResultsFound and can never be orphaned.
		self.sess().query(Ignore).filter(Ignore.user_id == user_id).filter(Ignore.ignored_user_id == unignore_user_id).delete()

	# --- offline direct messages (cache-free workers; see the offline_messages table) ---
	# All of these are ONE uncommitted transaction: _run_db owns the commit and retries the
	# whole unit on a serialization error, so they must not self-commit.

	def _user_id_bot_from_username(self, username):
		# cache-free resolve returning (id, bot). Mirrors _user_id_from_username but also
		# returns the bot flag, which the offline-message path needs to refuse bots.
		# db collation is case-insensitive, so re-check the case in python.
		row = self.sess().query(User.id, User.username, User.bot).filter(User.username == username).first()
		if not row or row[1] != username:
			return None
		return (row[0], row[2])

	def _bump_offline_tombstone(self, sender_id, recipient_id, when, count):
		# Upsert the single (sender, recipient) tombstone: one row stands for every message
		# from that sender that the recipient will never see, whether dropped by the cap or
		# expired by the TTL. Keeps the newest send time so the recipient can tell roughly
		# when they were being messaged.
		tomb = self.sess().query(OfflineMessage).filter(OfflineMessage.recipient_user_id == recipient_id).filter(OfflineMessage.sender_user_id == sender_id).filter(OfflineMessage.msg == None).first()
		if tomb is None:
			self.sess().add(OfflineMessage(sender_id, recipient_id, when, None, False, count))
			return
		tomb.dropped_count = (tomb.dropped_count or 0) + count
		if when > tomb.time:
			tomb.time = when

	def do_enqueue_offline_message(self, sender_id, recipient_username, msg, ex_msg):
		# ONE unit: resolve -> refuse bots -> honour the recipient's ignore list -> enforce
		# the per-pair cap -> insert. Returns a verdict tuple for the reactor callback.
		resolved = self._user_id_bot_from_username(recipient_username)
		if resolved is None:
			return ('nouser',)
		recipient_id, bot = resolved
		if bot:
			# Bots cannot receive offline messages at all: a queue of commands delivered in
			# one burst at login would flood an autohost, which is abuse waiting to happen.
			return ('bot',)
		if recipient_id == sender_id:
			return ('nouser',) # a user is never offline to themselves; nothing to queue
		if self.is_ignored(recipient_id, sender_id):
			# The recipient ignores the sender. Never write the row, and report success:
			# ignore status must not be observable by probing.
			return ('ok', recipient_id)
		now = datetime.now()
		n = self.sess().query(OfflineMessage).filter(OfflineMessage.recipient_user_id == recipient_id).filter(OfflineMessage.sender_user_id == sender_id).filter(OfflineMessage.msg != None).count()
		if n >= OFFLINE_MSG_MAX_PER_PAIR:
			# At the cap: drop the oldest content to make room, and account for it in the
			# tombstone so the loss is reported rather than silent.
			oldest = self.sess().query(OfflineMessage).filter(OfflineMessage.recipient_user_id == recipient_id).filter(OfflineMessage.sender_user_id == sender_id).filter(OfflineMessage.msg != None).order_by(OfflineMessage.id).first()
			dropped_time = oldest.time
			self.sess().delete(oldest)
			self.sess().flush() # the delete must land before the tombstone query re-reads
			self._bump_offline_tombstone(sender_id, recipient_id, dropped_time, 1)
		self.sess().add(OfflineMessage(sender_id, recipient_id, now, msg, ex_msg))
		return ('ok', recipient_id)

	def do_fetch_offline_messages(self, recipient_id):
		# Sender usernames are resolved in-query (like get_channel_messages) so the reactor
		# callback needs no DB of its own. Ordered by id: the send order matches the order
		# the messages were queued in.
		res = self.sess().query(OfflineMessage.id, OfflineMessage.sender_user_id, User.username, OfflineMessage.time, OfflineMessage.msg, OfflineMessage.ex_msg, OfflineMessage.dropped_count).filter(OfflineMessage.recipient_user_id == recipient_id).join(User, User.id == OfflineMessage.sender_user_id, isouter=True).order_by(OfflineMessage.id).all()
		return [(id, sender_id, username or "?", when, msg, ex_msg, dropped or 0)
		        for (id, sender_id, username, when, msg, ex_msg, dropped) in res]

	def do_delete_offline_messages(self, ids):
		if not ids:
			return 0
		return self.sess().query(OfflineMessage).filter(OfflineMessage.id.in_(ids)).delete(synchronize_session=False)

	def add_channel_message(self, channel_id, user_id, bridged_id, msg, ex_msg, date=None):
		if date is None:
			date = datetime.now()
		entry = ChannelHistory(channel_id, user_id, bridged_id, date, msg, ex_msg)
		self.sess().add(entry)
		self.sess().commit()
		return entry.id

	def get_channel_messages(self, user_id, channel_id, last_msg_id, limit):
		# returns (msgs, truncated) where msgs is [(date, username, msg, ex_msg, id), ...]
		# after last_msg_id, oldest-first. user_id is unused (kept: existing signature).
		#
		# Takes the NEWEST limit rows rather than the oldest: a client resuming from
		# last_msg_id=0 wants recent context, and handing it the oldest rows of the 14 day
		# retention window would be useless. A client resuming after a short disconnect is
		# under the cap either way, so the cap only bites on a cold start. Fetching limit+1
		# probes for more without a second COUNT: getting more than limit back means older
		# messages were elided, reported so the caller can say so instead of truncating silently.
		res = self.sess().query(ChannelHistory.time, ChannelHistory.msg, ChannelHistory.ex_msg, User.username, BridgedUser.external_username, BridgedUser.location, ChannelHistory.id).filter(ChannelHistory.channel_id == channel_id).filter(ChannelHistory.id > last_msg_id).join(User, isouter=True).join(BridgedUser, isouter=True).order_by(ChannelHistory.id.desc()).limit(limit + 1).all()
		truncated = len(res) > limit
		res = res[:limit]
		res.reverse()
		msgs = []
		for (htime, msg, ex_msg, username, external_username, location, id) in res:
			if not username:
				msgs.append((htime, "?", msg, ex_msg, id))
			elif external_username:
				bridged_username = external_username + ":" + location
				msgs.append((htime, bridged_username, msg, ex_msg, id))
			else:
				msgs.append((htime, username, msg, ex_msg, id))
		return (msgs, truncated)

class OfflineBridgedClient():
	def __init__(self, sqluser):
		# db fields
		self.bridged_id = sqluser.id
		self.location = sqluser.location
		self.external_id = sqluser.external_id
		self.external_username = sqluser.external_username
		self.last_bridged = sqluser.last_bridged

		# non-db fields
		self.username = self.external_username + ':' + self.location  
		self.channels = set()
		self.bridge_user_id = None

class BridgedUsersHandler:
	def __init__(self, root):
		self._root = root

	def sess(self):
		return self._root.session_manager.sess()

	def bridgedClient(self, location, external_id):
		entry = self.sess().query(BridgedUser).filter(BridgedUser.external_id == external_id).filter(BridgedUser.location == location).first()
		if not entry:
			return
		return OfflineBridgedClient(entry)

	def bridgedClientFromID(self, bridged_id):
		entry = self.sess().query(BridgedUser).filter(BridgedUser.id == bridged_id).first()
		if not entry:
			return
		return OfflineBridgedClient(entry)

	def bridgedClientFromUsername(self, username):
		external_username,location = username.split(':',1)
		if not external_username or not location:
			return
		entry = self.sess().query(BridgedUser).filter(BridgedUser.external_username == external_username).filter(BridgedUser.location == location).first()
		if not entry:
			return
		return OfflineBridgedClient(entry)

	def new_bridge_user(self, location, external_id, external_username):
		now = datetime.now()
		entry = BridgedUser(location, external_id, external_username, now)
		self.sess().add(entry)
		self.sess().commit()
		return entry

	def bridge_user(self, location, external_id, external_username):
		bridgedUser = self.sess().query(BridgedUser).filter(BridgedUser.external_id == external_id).filter(BridgedUser.location == location).first()
		entry = self.sess().query(BridgedUser).filter(BridgedUser.external_username == external_username).filter(BridgedUser.location == location).first()
		if (entry and entry.external_id != external_id):
			return False, "Another bridged user (external_id '%s') with location '%s' is currently associated to the external username '%s'" % (entry.external_id, location, external_username)
		if not bridgedUser:
			entry = self.new_bridge_user(location, external_id, external_username)
			return True, OfflineBridgedClient(entry)
		bridgedUser.external_username = external_username
		bridgedUser.last_bridged = datetime.now()
		self.sess().commit()
		return True, OfflineBridgedClient(bridgedUser)

	def clean(self):
		# remove any bridged user that wasn't seen for 30 days and isn't banned from any channels
		bridge_bans = self.sess().query(ChannelBridgedBan.bridged_id).distinct()
		ignore_ids = set(b.bridged_id for b in bridge_bans)		
		now = datetime.now()
		response = self.sess().query(BridgedUser).filter(BridgedUser.last_bridged < now - timedelta(days=30)).filter(not BridgedUser.id in ignore_ids)
		logging.info("deleting %i inactive bridged users", response.count())
		response.delete(synchronize_session=False)
		self.sess().commit()

class CachedBan(object):
	# session-independent snapshot of a Ban row, used by the 1.1 check_ban cache so
	# cached entries can't be invalidated by session expire/close (DetachedInstanceError).
	__slots__ = ('reason', 'end_date', 'user_id', 'ip', 'email')
	def __init__(self, ban):
		self.reason = ban.reason
		self.end_date = ban.end_date
		self.user_id = ban.user_id
		self.ip = ban.ip
		self.email = ban.email

class BansHandler:
	def __init__(self, root):
		self._root = root

		# 1.1: cache check_ban results. Bans change rarely but check_ban runs on every
		# login and registration. Keyed by (user_id, ip, email); stores the matched
		# CachedBan snapshot (or None) with a short TTL, so a newly-issued ban still
		# takes effect within the TTL even if no flush reached us, and every ban
		# add/remove flushes the cache immediately for instant effect.
		# 3.1: since login moved off the reactor thread, check_ban (and so this dict) is
		# now also read/written from deferToThread worker threads. No lock is used: dict
		# get/set are atomic under the GIL (as the codebase already relies on elsewhere,
		# e.g. nonres_registrations), and the cache is explicitly staleness-tolerant, so a
		# racing flush/write only costs a redundant query or a sub-TTL stale entry.
		self._ban_cache_ttl = 300  # seconds
		self._ban_cache = {}       # (user_id, ip, email) -> (CachedBan|None, expiry)

	def sess(self):
		return self._root.session_manager.sess()

	def _flush_ban_cache(self):
		self._ban_cache.clear()

	def check_ban(self, user_id=None, ip=None, email=None, now=None):
		# check if any of the args are currently banned
		if not now:
			now = datetime.now()
		key = (user_id, ip, email)
		hit = self._ban_cache.get(key)
		if hit is not None:
			result, expiry = hit
			# honour the TTL, and ensure a cached ban hasn't itself expired meanwhile
			if now <= expiry and (result is None or now <= result.end_date):
				return result
		result = self._query_ban(user_id, ip, email, now)
		self._ban_cache[key] = (result, now + timedelta(seconds=self._ban_cache_ttl))
		return result

	def _query_ban(self, user_id, ip, email, now):
		# fixme: its probably possible to do this with a single db command!
		if user_id:
			userban = self.sess().query(Ban).filter(Ban.user_id == user_id, now <= Ban.end_date).first()
			if userban:
				return CachedBan(userban)
		if ip:
			ipban = self.sess().query(Ban).filter(Ban.ip == ip, now <= Ban.end_date).first()
			if ipban:
				return CachedBan(ipban)
		if email:
			emailban = self.sess().query(Ban).filter(Ban.email == email, now <= Ban.end_date).first()
			if emailban:
				return CachedBan(emailban)
		return None

	def ban(self, issuer, duration, reason, username):
		# ban the user_id, current ip, and current email, of the target username (as a single ban)
		try:
			duration = float(duration)
		except:
			return False, 'Duration must be a float, cannot convert %s' % duration

		entry = self.sess().query(User).filter(User.username==username).first()
		if not entry:
			return False, "Unable to ban %s, user doesn't exist" % username
		ban = Ban(issuer.user_id, duration, reason, entry.id, entry.last_ip, entry.email)
		self.sess().add(ban)
		self.sess().commit()
		self._flush_ban_cache()
		return True, 'Successfully banned %s, %s, %s for %s days.' % (username, entry.last_ip, entry.email, duration)

	def ban_specific(self, issuer, duration, reason, arg):
		# arg might be username, ip or email; ban it
		try:
			duration = float(duration)
		except:
			return False, 'Duration must be a float, cannot convert %s' % duration

		email_match,_ = self._root.verificationdb.valid_email_addr(arg)
		ip_match = re.match(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b", arg)
		entry = self.sess().query(User).filter(User.username==arg).first()
		if email_match:
			ban = Ban(issuer.user_id, duration, reason, None, None, arg)
		elif ip_match:
			ban = Ban(issuer.user_id, duration, reason, None, arg, None)
		elif entry:
			ban = Ban(issuer.user_id, duration, reason, entry.id, None, None)
		else:
			return False, "Unable to match '%s' to username/ip/email" % arg
		self.sess().add(ban)
		self.sess().commit()
		self._flush_ban_cache()
		return True, 'Successfully banned %s for %s days' % (arg, duration)

	def unban(self, issuer, arg):
		# arg might be username, ip or email; remove all associated bans
		email_match,_ = self._root.verificationdb.valid_email_addr(arg)
		ip_match = re.match(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b", arg)
		entry = self.sess().query(User).filter(User.username==arg).first()
		if not (email_match or ip_match or entry):
			return False, "Unable to match '%s' to username/ip/email" % arg
		result = []
		n_unban = 0
		if email_match:
			results = self.sess().query(Ban).filter(Ban.email==arg)
			for result in results:
				self.sess().delete(result)
				n_unban += 1
		if ip_match:
			results = self.sess().query(Ban).filter(Ban.ip==arg)
			for result in results:
				self.sess().delete(result)
				n_unban += 1
		if entry:
			results = self.sess().query(Ban).filter(Ban.user_id==entry.id)
			for result in results:
				self.sess().delete(result)
				n_unban += 1
		self.sess().commit()
		self._flush_ban_cache()
		if n_unban>0:
			return True, 'Successfully removed %s bans relating to %s' % (n_unban, arg)
		else:
			return False, 'No matching bans for %s' % arg

	def check_blacklist(self, email):
		partition = email.partition('@')
		if partition[1]=="":
			return None
		domain = partition[2]
		entry = self.sess().query(BlacklistedEmailDomain).filter(BlacklistedEmailDomain.domain==domain).first()
		if entry:
			return entry
		return None

	def blacklist(self, issuer, domain, reason):
		if not '.' in domain:
			return False, "invalid domain '%s', contains no '.'" % domain
		if 'www' in domain or 'http' in domain or '/' in domain:
			return False, "invalid domain '%s', do not include www or http(s) part, example: hawtmail.com" % domain
		entry = self.sess().query(BlacklistedEmailDomain).filter(BlacklistedEmailDomain.domain==domain).first()
		if entry:
			return False, 'Domain %s is already blacklisted' % domain
		entry = BlacklistedEmailDomain(issuer.user_id, domain, reason)
		self.sess().add(entry)
		self.sess().commit()
		return True, 'Successfully added %s to blacklist' % domain

	def unblacklist(self, issuer, domain):
		entry = self.sess().query(BlacklistedEmailDomain).filter(BlacklistedEmailDomain.domain==domain).first()
		if not entry:
			return False, "Unable to remove %s, entry doesn't exist" % (domain)
		self.sess().delete(entry)
		self.sess().commit()
		return True, "Sucessfully removed %s from blacklist" % domain

	def list_bans(self):
		# return a list of all bans
		banlist = []
		for ban in self.sess().query(Ban):
			username = None
			issuer = None
			if ban.user_id:
				entry = self.sess().query(User).filter(User.id==ban.user_id).first()
				if entry: username = entry.username
			if ban.issuer_user_id:
				issuer =  self.sess().query(User).filter(User.id==ban.issuer_user_id).first()
				if issuer: issuer_username = issuer.username
			banlist.append({
				'username': username or "",
				'id': ban.user_id,
				'ip': ban.ip or "",
				'email': ban.email or "",
				'end_date': ban.end_date.strftime("%Y-%m-%d %H:%M"),
				'reason': ban.reason,
				'issuer': issuer_username
			})
		return banlist

	def list_blacklist(self):
		# return a list of all blacklisted email domains
		blacklist = []
		for item in self.sess().query(BlacklistedEmailDomain):
			issuer =  self.sess().query(User).filter(User.id==item.issuer_user_id).first()
			if issuer: issuer_username = issuer.username
			blacklist.append({
				'domain': item.domain,
				'start_time': item.start_time.strftime("%Y-%m-%d %H:%M"),
				'reason': item.reason or "",
				'issuer': issuer_username
			})
		return blacklist

	def clean(self):
		# remove all expired bans
		now = datetime.now()
		response = self.sess().query(Ban).filter(Ban.end_date < now)
		logging.info("deleting %i expired bans", response.count())
		response.delete(synchronize_session=False)
		self.sess().commit()
		self._flush_ban_cache()

class VerificationsHandler:
	def __init__(self, root):
		self._root = root
		self.require_verification = (self._root.mail_user != None)	
		self.mail_identity = getattr(self._root, "mail_identity", "Recoil Engine")
		self.mail_contact_addr = getattr(self._root, "mail_contact_addr", "https://recoilengine.org")
		
	def sess(self):
		return self._root.session_manager.sess()

	def active(self):
		return self.require_verification

	def valid_email_addr(self, email):
		assert(type(email) == str)
		if (not email or email==""):
			return False, "An email address is required."
		if ' ' in email:
			return False, "Invalid email address (check for whitespace)."
		if not re.match(r"[a-z0-9._%+-]+@[a-z0-9.-]+\.[a-z]{2,6}", email):
			return False, "Invalid email address format."
		return True, ""

	def check_and_send(self, user_id, email, digits, reason):
		# check that we don't already have an active verification, send a new one if not
		if not self.active():
			return True, ''
		good, validity_reason = self.valid_email_addr(email)
		if not good:
			return False, validity_reason
		dbblacklist = self._root.bandb.check_blacklist(email)
		if dbblacklist:
			return False, dbblacklist.domain + " is blacklisted: " + dbblacklist.reason
		# no need to check if ip is banned, that is done by login!

		email_entry = self.sess().query(Verification).filter(Verification.email == email).first()
		if email_entry:
			if datetime.now() <= email_entry.expiry:
				return False, 'A verification attempt is already active for ' + email + ', use that or wait for it to expire (up to 24h)'
		if email_entry: #expired
			self.remove(email_entry.user_id)

		entry = self.sess().query(Verification).filter(Verification.user_id == user_id).first()
		if entry:
			if entry.email != email:
				return False, 'A verification code is active for ' + entry.email + ', use that or wait for it to expire (up to 24h)'
			if datetime.now() < entry.expiry:
				return False, 'Already sent a verification code, please check your spam filter!'
		if entry: #expired
			self.remove(user_id)
		entry = self.create(user_id, email, digits, reason)
		self.send(entry)
		return True, ''

	def create(self, user_id, email, digits, reason):
		entry = Verification(user_id, email, digits, reason)
		self.sess().add(entry)
		self.sess().commit()
		return entry

	def resend(self, user_id, email):
		entry = self.sess().query(Verification).filter(Verification.user_id == user_id).first()
		if not entry:
			return False, 'You do not have an active verification code'
		if entry.expiry <= datetime.now():
			return False, 'Your verification code has expired, please request a new one'
		if email!=entry.email:
			return False, 'Your verification code for ' + entry.email + ' cannot be re-sent to a different email address, use it or wait for it to expire (up to 48h)'
		if entry.resends>=3:
			return False, 'Too many resends, please try again later'
		if entry.resends==0:
			entry.reason += " (resend requested)"
		entry.resends += 1
		self.sess().commit()
		self.send(entry)
		return True, ''

	def send(self, entry):
		if not self.active(): 
			return
		try:
			thread.start_new_thread(self._send, (entry.email, entry.code, entry.reason, entry.expiry))
		except:
			logging.error('Failed to launch VerificationHandler._send: %s, %s, %s' % (entry, entry.reason, entry.expiry))
			return
		dbuser = self.sess().query(User).filter(User.id == entry.user_id).first()
		logging.info('Sent verification code for <%s> to %s' % (dbuser.username, entry.email)) 

	def _send(self, email, code, reason, expiry):
		sent_from = self._root.mail_user
		to = email
		tz_label = getattr(self._root, "mail_tz_label", "UTC")
		subject = getattr(self._root, "mail_subject", self.mail_identity + " verification code")
		body_template = getattr(self._root, "mail_body_template", None)
		if body_template:
			body = body_template.format(
				name=self.mail_identity,
				reason=reason,
				code=str(code),
				expiry_date=expiry.strftime('%Y-%m-%d'),
				expiry_time=expiry.strftime('%H:%M'),
				tz=tz_label,
				contact=self.mail_contact_addr,
				username=reason.split('(username: ')[-1].rstrip(')') if '(username:' in reason else ''
			)
		else:
			body = "You are receiving this email because you recently " + reason + ".\r\nYour email verification code is " + str(code) + "\r\n\r\nThis verification code will expire on " + expiry.strftime("%Y-%m-%d") + " at " + expiry.strftime("%H:%M") + " " + tz_label + "."
		self._send_email(sent_from, to, subject, body)


	def _send_email(self, sent_from, to, subject, body):
		if not self.active(): #safety
			logging.error("Attempt to _send_email (subject: %s) failed, verifications handler is inactive" % subject)
			return
		if not getattr(self._root, "mail_body_template", None):
			body += "\r\n\r\nIf you received this message in error, please contact us at " + self.mail_contact_addr + ". Direct replies to this message will be automatically deleted."
		message = MIMEText(body, 'plain')
		message['Subject'] = subject
		message['From'] = self.mail_identity + " <" + sent_from + ">"
		message['To'] = to
		try:
			smtp_host = getattr(self._root, 'mail_smtp_host', None)
			smtp_port = int(getattr(self._root, 'mail_smtp_port', 587))
			smtp_user = getattr(self._root, 'mail_smtp_user', None)
			smtp_pass = getattr(self._root, 'mail_smtp_pass', None)
			if smtp_host:
				server = smtplib.SMTP(smtp_host, smtp_port)
				server.ehlo()
				server.starttls()
				server.ehlo()
				if smtp_user and smtp_pass:
					server.login(smtp_user, smtp_pass)
			else:
				server = smtplib.SMTP()
				server.connect()
			server.sendmail(sent_from, to, message.as_string())
			server.close()
			logging.info('Sent email to %s' % (to))
		except Exception as e:
			logging.error('Failed to send email from %s to %s' % (sent_from, to))
			logging.error(str(e))

	def verify (self, user_id, email, code):
		if not self.active():
			return True, ''
		if code=="":
			return False, 'A verification code is required -- check your email'
		entry = self.sess().query(Verification).filter(Verification.user_id == user_id).first() # there should be (at most) one code per user
		if not entry:
			logging.error('Unexpected verification attempt: %s, %s' % (user_id, code))
			return False, 'Unexpected verification attempt, please request a verification code'
		if entry.expiry <= datetime.now():
			return False, 'Your verification code for ' + entry.email + ' has expired, please request a new one'
		if entry.attempts>=3:
			return False, 'Too many attempts, please try again later'
		if entry.email!=email:
			return False, 'Failed to match email addresses'
		try:
			if entry.code==int(code):
				dbuser = self.sess().query(User).filter(User.id == user_id).first()
				logging.info('Successful verification code for <%s> %s' % (dbuser.username, entry.email)) 
				self.remove(user_id)
				return True, ''
			else:
				entry.attempts += 1
				self.sess().commit()
				return False, 'Incorrect verification code, %i/3 attempts remaining' % (3-entry.attempts)
		except Exception as e:
			entry.attempts += 1
			self.sess().commit()
			return False, 'Incorrect verification code, ' + str(e) + ', %i/3 attempts remaining' % (3-entry.attempts)

	def remove(self, user_id):
		# remove all entries for user
		self.sess().query(Verification).filter(Verification.user_id == user_id).delete(synchronize_session=False)
		self.sess().commit()

	def clean(self):
		# remove all expired entries
		now = datetime.now()
		response = self.sess().query(Verification).filter(Verification.expiry < now)
		logging.info("deleting %i expired verifications", response.count())
		response.delete(synchronize_session=False)
		self.sess().commit()

	def reset_password(self, user_id, email_to_user):
		# reset pw, email to user
		char_set = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890!£$%^&*?"
		new_password_raw = ""
		for i in range(0,10):
			new_password_raw += random.choice(char_set)
		hash = hashlib.md5()
		hash.update(str.encode(new_password_raw))
		new_password = base64.b64encode(hash.digest()).decode()
		assert(self._root.protocol._validPasswordSyntax(new_password))

		dbuser = self.sess().query(User).filter(User.id == user_id).first()
		dbuser.password = new_password
		uname = dbuser.username
		self.sess().commit()
		self._root.userdb.invalidate_user_cache(user_id, uname)

		if email_to_user:
			try:
				thread.start_new_thread(self._send_reset_password_email, (dbuser.email, dbuser.username, new_password_raw,))
			except:
				logging.error('Failed to launch VerificationsHandler._send_reset_password_email: %s' % (dbuser))

	def _send_reset_password_email(self, email, username, password):
		sent_from = self._root.mail_user
		to = email
		subject = getattr(self._root, 'mail_subject_recovery', self.mail_identity + ' account recovery')
		body = "You are receiving this email because you recently requested to recover the account <" + username + "> at the " + self.mail_identity + " lobby server.\r\nYour new password is " + password
		self._send_email(sent_from, to, subject, body)


class ChannelsHandler:
	def __init__(self, root):
		self._root = root

	def sess(self):
		return self._root.session_manager.sess()

	def channel_from_name(self, name):
		entry = self.sess().query(Channel).filter(Channel.name == name).first()
		return entry

	def channel_from_id(self, channel_id):
		entry = self.sess().query(Channel).filter(Channel.id == channel_id).first()
		return entry

	def all_channels(self):
		response = self.sess().query(Channel)
		channels = {}
		for chan in response:
			channels[chan.name] = {
					'id': chan.id,
					'owner_user_id':chan.owner_user_id,
					'key':chan.key,
					'topic':chan.topic or '',
					'topic_user_id':chan.topic_user_id,
					'antispam':chan.antispam,
					'operator':[],
					'chanserv': True,
					'store_history': chan.store_history,
					'last_used' : chan.last_used,
				}
		return channels

	def all_operators(self):
		response = self.sess().query(ChannelOp)
		operators = []
		for op in response:
			operators.append({
					'channel_id': op.channel_id,
					'user_id': op.user_id,
				})
		return operators

	def all_bans(self):
		response = self.sess().query(ChannelBan)
		bans = []
		for ban in response:
			bans.append({
					'channel_id': ban.channel_id,
					'issuer_user_id': ban.issuer_user_id,
					'user_id': ban.user_id,
					'ip_address': ban.ip_address,
					'expires': ban.expires,
					'reason': ban.reason,
				})
		return bans

	def all_bridged_bans(self):
		response = self.sess().query(ChannelBridgedBan)
		bans = []
		for ban in response:
			bans.append({
					'channel_id': ban.channel_id,
					'bridged_id': ban.bridged_id,
					'issuer_user_id': ban.issuer_user_id,
					'expires': ban.expires,
					'reason': ban.reason,
				})
		return bans

	def all_mutes(self):
		response = self.sess().query(ChannelMute)
		mutes = []
		for mute in response:
			mutes.append({
					'channel_id': mute.channel_id,
					'issuer_user_id': mute.issuer_user_id,
					'user_id': mute.user_id,
					'expires': mute.expires,
					'reason': mute.reason,
				})
		return mutes

	def all_forwards(self):
		response = self.sess().query(ChannelForward)
		forwards = []
		for forward in response:
			forwards.append({
			'channel_from_id': forward.channel_from_id,
			'channel_to_id': forward.channel_to_id,
			})
		return forwards
		
	def recordUse(self, channel):
		now = datetime.now()
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		if entry:
			now = datetime.now()
			entry.last_used = now
			self.sess().commit()

	def setTopic(self, channel, topic, target):
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		if entry:
			entry.topic = topic
			entry.topic_user_id = target.user_id
			self.sess().commit()

	def setKey(self, channel, key):
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		if entry:
			entry.key = key
			self.sess().commit()

	def setFounder(self, channel, target):
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		if entry:
			entry.owner_user_id = target.user_id
			self.sess().commit()

	def setAntispam(self, channel, antispam):
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		if entry:
			entry.antispam = antispam
			self.sess().commit()

	def opUser(self, channel, target):
		entry = ChannelOp(channel.id, target.user_id)
		self.sess().add(entry)
		self.sess().commit()

	def deopUser(self, channel, target):
		response = self.sess().query(ChannelOp).filter(ChannelOp.user_id == target.user_id).filter(ChannelOp.channel_id == channel.id)
		response.delete()
		self.sess().commit()

	def banBridgedUser(self, channel, issuer, target, expires, reason):
		entry = ChannelBridgedBan(channel.id, issuer.user_id, target.bridged_id, expires, reason)
		self.sess().add(entry)
		self.sess().commit()

	def unbanBridgedUser(self, channel, target):
		response = self.sess().query(ChannelBridgedBan).filter(ChannelBridgedBan.bridged_id == target.bridged_id).filter(ChannelBridgedBan.channel_id == channel.id)
		response.delete()
		self.sess().commit()

	def banUser(self, channel, issuer, target, expires, reason):
		entry = ChannelBan(channel.id, issuer.user_id, target.user_id, target.last_ip, expires, reason)
		self.sess().add(entry)
		self.sess().commit()

	def unbanUser(self, channel, target):
		response = self.sess().query(ChannelBan).filter(ChannelBan.user_id == target.user_id).filter(ChannelBan.channel_id == channel.id)
		response.delete()
		self.sess().commit()

	def muteUser(self, channel, issuer, target, expires, reason):
		entry = ChannelMute(channel.id, issuer.user_id, target.user_id, expires, reason)
		self.sess().add(entry)
		self.sess().commit()

	def unmuteUser(self, channel, target):
		response = self.sess().query(ChannelMute).filter(ChannelMute.user_id == target.user_id).filter(ChannelMute.channel_id == channel.id)
		response.delete()
		self.sess().commit()

	def setHistory(self, chan, enable):
		entry = self.sess().query(Channel).filter(Channel.name == chan.name).first()
		if entry:
			entry.store_history = enable
			self.sess().commit()

	def addForward(self, channel_from, channel_to):
		entry = ChannelForward(channel_from.id, channel_to.id)
		self.sess().add(entry)
		self.sess().commit()

	def removeForward(self, channel_from, channel_to):
		response = self.sess().query(ChannelForward).filter(ChannelForward.channel_from_id == channel_from.id).filter(ChannelForward.channel_to_id == channel_to.id)
		response.delete()
		self.sess().commit()

	def register(self, channel, target):
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		if not entry:
			entry = Channel(channel.name)
		if channel.topic:
			entry.topic = channel.topic
			entry.topic_user_id = target.user_id
		entry.owner_user_id = target.user_id
		entry.last_used = datetime.now()
		self.sess().add(entry)
		self.sess().commit()
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		channel.id = entry.id

	def unRegister(self, channel):
		self.sess().query(Channel).filter(Channel.name == channel.name).delete()
		self.sess().commit()

	def registered(self, channel):
		entry = self.sess().query(Channel).filter(Channel.name == channel.name).first()
		return bool(entry)
		
	def clean(self):
		#delete all expired channel bans/mutes:
		now = datetime.now()

		response = self.sess().query(ChannelMute).filter(ChannelMute.expires < now)
		logging.info("deleting %i expired channel mutes", response.count())
		response.delete(synchronize_session=False)

		response = self.sess().query(ChannelBan).filter(ChannelBan.expires < now)
		logging.info("deleting %i expired channel bans", response.count())
		response.delete(synchronize_session=False)

		response = self.sess().query(ChannelBridgedBan).filter(ChannelBridgedBan.expires < now)
		logging.info("deleting %i expired channel bridged bans", response.count())
		response.delete(synchronize_session=False)

		response = self.sess().query(Channel).filter(Channel.last_used < now - timedelta(days=180))
		logging.info("deleting %i inactive channels", response.count())
		response.delete(synchronize_session=False)
		
		self.sess().commit()


class ContentHandler:
	def __init__(self, root):
		self._root = root

	def sess(self):
		return self._root.session_manager.sess()

	def set_min_spring_version(self, version):
		response = self.sess().query(MinSpringVersion)
		response.delete()
		now = datetime.now()
		entry = MinSpringVersion(version, now)
		self.sess().add(entry)
		self.sess().commit()	
	
	def get_min_spring_version(self):
		entry = self.sess().query(MinSpringVersion).first()
		if entry: return entry.min_spring_version
		return '*'


if __name__ == '__main__':
	class root:
		censor = False

	import sqlalchemy, os
	try: # cleanup old db
		os.remove("test.db")
	except:
		pass
	engine = sqlalchemy.create_engine("sqlite:///test.db", echo=False)
	def _fk_pragma_on_connect(dbapi_con, con_record):
	        dbapi_con.execute('PRAGMA journal_mode = MEMORY')
	        dbapi_con.execute('PRAGMA synchronous = OFF')
	sqlalchemy.event.listen(engine, 'connect', _fk_pragma_on_connect)

	root = root()
	root.session_manager = session_manager(root, engine)
	root.mail_user = "no-reply@recoilengine.org"
	
	userdb = UsersHandler(root)
	channeldb = ChannelsHandler(root)
	verificationdb = VerificationsHandler(root)
	bandb = BansHandler(root)
	root.userdb = userdb
	root.channeldb = channeldb
	root.verificationdb = verificationdb
	root.bandb = bandb

	# test save/load user
	username = u"test"
	userdb.register_user(username, u"pass", "192.168.1.1", "blackhole@blackhole.io")
	client = userdb.clientFromUsername(username)
	assert(isinstance(client.id, int))

	# test verification
	entry = verificationdb.create(client.id, client.email, 4, "test")
	verificationdb._send_email("test@test.test", "blackhole2@blackhole.io", "test", "test") #use main thread, or Python will exit without waiting for the test!
	verificationdb.verify(client.id, client.email, entry.code)
	verificationdb.clean()

	# test ban/unban
	client.user_id = client.id # ban issuer is an *online* client; impersonate one
	userdb.register_user("delinquent", u"pass", "192.168.1.2", "blackhole3@blackhole.io")
	client2 = userdb.clientFromUsername("delinquent")
	bandb.ban(client, 1, "test", "delinquent")
	ban = bandb.check_ban(client2.id, None, None)
	assert(ban)
	bandb.unban(client, "delinquent")
	ban = bandb.check_ban(client2.id, None, None)
	assert(not ban)

	# test save/load channel
	channelname = u"testchannel"
	channel = Channel(channelname)
	client.user_id = client.id # only online clients can register channels, so pretend we are one of those
	channeldb.register(channel, client)
	assert(channel.id > 0)

	# test setHistory
	assert(channel.store_history == False)
	channel.store_history = True
	channeldb.setHistory(channel, channel.store_history)
	channel = channeldb.channel_from_name(channelname)
	assert(channel.store_history == True)

	# test channel message history
	now = datetime.now()
	msg = u'test message %d äöüÄÖÜß ?(?_°)?'
	last_msg_id = -1
	for i in range(0, 20):
		if i == 0:
			last_msg_id = userdb.add_channel_message(channel.id, client.id, None, msg % i, False, now + timedelta(0, i))
		else:
			userdb.add_channel_message(channel.id, client.id, None, msg % i, False, now + timedelta(0, i))

	assert(last_msg_id > -1)

	for i in range(0,21):
		# NB: user_id/channel_id were passed the wrong way round here before the limit was
		# added. It passed only because user_id is unused and a fresh db gives the first
		# user and the first channel the same id, so the channel_id filter matched anyway.
		msgs, truncated = userdb.get_channel_messages(client.id, channel.id, last_msg_id + i -1, 200)
		assert(len(msgs) == 20 - i)
		assert(truncated == False)
		if (len(msgs) > 0):
			assert(msgs[0][0] == now + timedelta(0, i))
			assert(msgs[0][1] == client.username)
			assert(msgs[0][2] == msg % i)
			assert(type(msgs[0][2]) == str)

	# the read cap returns the NEWEST messages, oldest-first, and reports the truncation
	msgs, truncated = userdb.get_channel_messages(client.id, channel.id, last_msg_id - 1, 5)
	assert(truncated == True)
	assert(len(msgs) == 5)
	assert(msgs[0][2] == msg % 15) # newest 5 of 20, not the oldest 5
	assert(msgs[4][2] == msg % 19)

	# a cap exactly matching the available rows is not a truncation
	msgs, truncated = userdb.get_channel_messages(client.id, channel.id, last_msg_id - 1, 20)
	assert(len(msgs) == 20)
	assert(truncated == False)

	userdb.add_channel_message(channel.id, None, None, "test", False)
	userdb.add_channel_message(channel.id, 99, None, "test", False)
	userdb.add_channel_message(channel.id, 99, 99, "test", False)

	userdb.clean()
	verificationdb.clean()
	bandb.clean()

	print("Tests went ok")