oedb-backend/oedb/resources/event.py

"""
Event resource for the OpenEventDatabase.
"""

import json
import re
import falcon
import psycopg2
import psycopg2.extras
import psycopg2.errors
from oedb.models.event import BaseEvent
from oedb.utils.db import db_connect
from oedb.utils.serialization import dumps
from oedb.utils.logging import logger

class EventResource(BaseEvent):
    """
    Resource for managing events.
    Handles the /event and /event/{id} endpoints.
    """
    
    def maybe_insert_geometry(self, geometry, cur):
        """
        Insert a geometry into the geo table if it doesn't exist.
        
        Args:
            geometry: The GeoJSON geometry.
            cur: The database cursor.
            
        Returns:
            tuple: The hash of the geometry.
        """
        logger.debug("Inserting geometry if not exists")
        # Insert into geo table if not existing
        cur.execute("""INSERT INTO geo
                            SELECT geom, md5(st_astext(geom)) as hash, st_centroid(geom) as geom_center FROM
                                    (SELECT st_setsrid(st_geomfromgeojson( %s ),4326) as geom) as g
                                WHERE ST_IsValid(geom)
                        ON CONFLICT DO NOTHING RETURNING hash;""",
                    (geometry,))
        # Get its id (md5 hash)
        h = cur.fetchone()
        if h is None:
            cur.execute("""SELECT md5(st_asewkt(geom)),
                            ST_IsValid(geom),
                            ST_IsValidReason(geom) from (SELECT st_geomfromgeojson( %s ) as geom) as g ;""", (geometry,))
            h = cur.fetchone()
        return h

    def relative_time(self, when, cur):
        """
        Convert a relative time string to database-friendly format.
        
        Args:
            when: The relative time string.
            cur: The database cursor.
            
        Returns:
            tuple: The start and stop times.
        """
        logger.debug(f"Converting relative time: {when}")
        when = when.upper().replace(' ', '+')
        event_start = cur.mogrify("%s", (when,)).decode("utf-8")
        event_stop = cur.mogrify("%s", (when,)).decode("utf-8")

        if when == 'NOW':
            event_start = "now()"
            event_stop = "now()"
        if when == 'TODAY':
            event_start = "CURRENT_DATE"
            event_stop = "CURRENT_DATE + INTERVAL '1 DAY'"
        if when == 'TOMORROW':
            event_start = "CURRENT_DATE + INTERVAL '1 DAY'"
            event_stop = "CURRENT_DATE + INTERVAL '2 DAY'"
        if when == 'YESTERDAY':
            event_start = "CURRENT_DATE - INTERVAL '1 DAY'"
            event_stop = "CURRENT_DATE"
        m = re.match('(LAST|NEXT)(YEAR|MONTH|WEEK|DAY|HOUR|MINUTE)', when)
        if m is not None:
            when = m.group(1) + '1' + m.group(2) + 'S'
        m = re.match('(LAST|NEXT)([0-9]*)(YEAR|MONTH|WEEK|MINUTE|HOUR|DAY)S', when)
        if m is not None:
            if m.group(1) == 'LAST':
                event_start = "now() - INTERVAL '%s %s'" % m.group(2, 3)
                event_stop = "now()"
            else:
                event_start = "now()"
                event_stop = "now() + INTERVAL '%s %s'" % m.group(2, 3)

        return event_start, event_stop

    def on_get(self, req, resp, id=None, geom=None):
        """
        Handle GET requests to the /event and /event/{id} endpoints.
        
        Args:
            req: The request object.
            resp: The response object.
            id: The event ID (optional).
            geom: The geometry to search with (optional).
        """
        logger.info(f"Processing GET request to /event{f'/{id}' if id else ''}")
        
        db = db_connect()
        cur = db.cursor(cursor_factory=psycopg2.extras.DictCursor)
        
        try:
            if id is None:
                event_sort = "createdate DESC"
                # Get query search parameters
                if geom is not None:
                    # Convert our geojson geom to WKT
                    geoj = json.dumps(geom)
                    # Buffer around geom?
                    if 'buffer' in req.params:
                        buffer = float(req.params['buffer'])
                    elif geom['type'] == 'Linestring':
                        buffer = 1000  # 1km buffer by default around Linestrings
                    else:
                        buffer = 0
                    if buffer == 0:
                        event_bbox = cur.mogrify(" AND ST_Intersects(geom, ST_SetSRID(ST_GeomFromGeoJSON(%s),4326)) ", (geoj,)).decode("utf-8")
                    else:
                        event_bbox = cur.mogrify(" AND ST_Intersects(geom, ST_Buffer(ST_SetSRID(ST_GeomFromGeoJSON(%s),4326)::geography, %s)::geometry) ", (geoj, buffer)).decode("utf-8")
                    event_dist = cur.mogrify("ST_Length(ST_ShortestLine(geom, ST_SetSRID(ST_GeomFromGeoJSON(%s),4326))::geography)::integer as distance, ", (geoj,)).decode("utf-8")
                    event_sort = cur.mogrify("ST_Length(ST_ShortestLine(geom, ST_SetSRID(ST_GeomFromGeoJSON(%s),4326))::geography)::integer, ", (geoj,)).decode("utf-8") + event_sort
                elif 'bbox' in req.params:
                    # Limit search with bbox (E,S,W,N)
                    event_bbox = cur.mogrify(" AND geom && ST_SetSRID(ST_MakeBox2D(ST_Point(%s,%s),ST_Point(%s,%s)),4326) ", tuple(req.params['bbox'])).decode("utf-8")
                    event_dist = ""
                elif 'near' in req.params:
                    # Limit search with location+distance
                    # (long, lat, distance in meters)
                    if len(req.params['near']) < 3:
                        dist = 1
                    else:
                        dist = req.params['near'][2]
                    event_bbox = cur.mogrify(" AND ST_Intersects(geom, ST_Buffer(st_setsrid(st_makepoint(%s,%s),4326)::geography,%s)::geometry) ", (req.params['near'][0], req.params['near'][1], dist)).decode("utf-8")
                    event_dist = cur.mogrify("ST_Length(ST_ShortestLine(geom, st_setsrid(st_makepoint(%s,%s),4326))::geography)::integer as distance,", (req.params['near'][0], req.params['near'][1])).decode("utf-8")
                    event_sort = cur.mogrify("ST_Length(ST_ShortestLine(geom, st_setsrid(st_makepoint(%s,%s),4326))::geography)::integer, ", (req.params['near'][0], req.params['near'][1])).decode("utf-8") + event_sort
                elif 'polyline' in req.params:
                    # Use encoded polyline as search geometry
                    if 'buffer' in req.params:
                        buffer = float(req.params['buffer'])
                    else:
                        buffer = 1000
                    if 'polyline_precision' in req.params:
                        precision = int(req.params['polyline_precision'])
                    else:
                        precision = 5
                    # ST_Scale is a workaround to postgis bug not taking precision into account in ST_LineFromEncodedPolyline
                    event_bbox = cur.mogrify(" AND ST_Intersects(geom, ST_Buffer(ST_Scale(ST_LineFromEncodedPolyline(%s),1/10^(%s-5),1/10^(%s-5))::geography, %s)::geometry) ", (req.params['polyline'], precision, precision, buffer)).decode("utf-8")
                    event_dist = cur.mogrify("ST_Length(ST_ShortestLine(geom, ST_Scale(ST_LineFromEncodedPolyline(%s),1/10^(%s-5),1/10^(%s-5)))::geography)::integer as distance, ", (req.params['polyline'], precision, precision)).decode("utf-8")
                elif 'where:osm' in req.params:
                    event_bbox = cur.mogrify(" AND events_tags ? 'where:osm' AND events_tags->>'where:osm'=%s ", (req.params['where:osm'],)).decode("utf-8")
                    event_dist = ""
                elif 'where:wikidata' in req.params:
                    event_bbox = cur.mogrify(" AND events_tags ? 'where:wikidata' AND events_tags->>'where:wikidata'=%s ", (req.params['where:wikidata'],)).decode("utf-8")
                    event_dist = ""
                else:
                    event_bbox = ""
                    event_dist = ""

                if 'when' in req.params:
                    # Limit search with fixed time
                    when = req.params['when'].upper()
                    event_when = "tstzrange(%s,%s,'[]')" % (self.relative_time(when, cur))
                elif 'start' in req.params and 'stop' in req.params:
                    # Limit search with fixed time (start to stop)
                    event_start, unused = self.relative_time(req.params['start'], cur)
                    unused, event_stop = self.relative_time(req.params['stop'], cur)
                    event_when = "tstzrange(%s,%s,'[]')" % (event_start, event_stop)
                elif 'start' in req.params and 'stop' not in req.params:
                    # Limit search with fixed time (start to now)
                    event_start, unused = self.relative_time(req.params['start'], cur)
                    event_when = "tstzrange(%s,now(),'[]')" % event_start
                elif 'start' not in req.params and 'stop' in req.params:
                    # Limit search with fixed time (now to stop)
                    unused, event_stop = self.relative_time(req.params['stop'], cur)
                    event_when = "tstzrange(now(),%s,'[]')" % event_stop
                else:
                    # Return events that are currently active (current time is between start and stop)
                    event_when = "now()"

                if 'what' in req.params:
                    # Limit search based on "what"
                    event_what = cur.mogrify(" AND events_what LIKE %s AND events_what LIKE %s ", (req.params['what'][:4] + "%", req.params['what'] + "%")).decode("utf-8")
                else:
                    event_what = ""

                if 'type' in req.params:
                    # Limit search based on type (scheduled, forecast, unscheduled)
                    event_type = cur.mogrify(" AND events_type = %s ", (req.params['type'],)).decode("utf-8")
                else:
                    event_type = ""

                if 'limit' in req.params:
                    limit = cur.mogrify("LIMIT %s", (req.params['limit'],)).decode("utf-8")
                else:
                    limit = "LIMIT 200"

                event_geom = "geom_center"
                geom_only = False
                if 'geom' in req.params:
                    if req.params['geom'] == 'full':
                        event_geom = "geom"
                    elif req.params['geom'] == 'only':
                        geom_only = True
                    else:
                        event_geom = cur.mogrify("ST_SnapToGrid(geom,%s)", (req.params['geom'],)).decode("utf-8")

                # Search recent active events.
                if event_when == "now()":
                    # Use @> operator to check if events_when contains current time
                    sql = """SELECT events_id, events_tags, createdate, lastupdate, {event_dist} st_asgeojson({event_geom}) as geometry, st_x(geom_center) as lon, st_y(geom_center) as lat
                                FROM events JOIN geo ON (hash=events_geo)
                                WHERE events_when @> {event_when} {event_what} {event_type} {event_bbox}
                                ORDER BY {event_sort} {limit}"""
                else:
                    # Use && operator to check if events_when overlaps with event_when
                    sql = """SELECT events_id, events_tags, createdate, lastupdate, {event_dist} st_asgeojson({event_geom}) as geometry, st_x(geom_center) as lon, st_y(geom_center) as lat
                                FROM events JOIN geo ON (hash=events_geo)
                                WHERE events_when && {event_when} {event_what} {event_type} {event_bbox}
                                ORDER BY {event_sort} {limit}"""
                # No user generated content here, so format is safe.
                sql = sql.format(event_dist=event_dist, event_geom=event_geom,
                                event_bbox=event_bbox, event_what=event_what,
                                event_when=event_when, event_type=event_type,
                                event_sort=event_sort, limit=limit)
                logger.debug(f"Executing SQL: {sql}")
                cur.execute(sql)
                resp.text = dumps(self.rows_to_collection(cur.fetchall(), geom_only))
                resp.status = falcon.HTTP_200
                logger.success(f"Successfully processed GET request to /event")
            else:
                # Get single event geojson Feature by id.
                logger.debug(f"Fetching event with ID: {id}")
                cur.execute("SELECT events_id, events_tags, createdate, lastupdate, st_asgeojson(geom) as geometry, st_x(geom_center) as lon, st_y(geom_center) as lat FROM events JOIN geo ON (hash=events_geo) WHERE events_id=%s", [id])

                e = cur.fetchone()
                if e is not None:
                    resp.text = dumps(self.row_to_feature(e))
                    resp.status = falcon.HTTP_200
                    logger.success(f"Successfully processed GET request to /event/{id}")
                else:
                    resp.status = falcon.HTTP_404
                    logger.warning(f"Event not found: {id}")
        except psycopg2.errors.InsufficientPrivilege as e:
            logger.error(f"Permission denied for database table: {e}")
            resp.status = falcon.HTTP_500
            resp.text = dumps({
                "error": "Database permission error",
                "message": "The server does not have permission to access the required database tables. Please contact the administrator to fix this issue.",
                "details": str(e)
            })
        except Exception as e:
            logger.error(f"Error processing GET request: {e}")
            resp.status = falcon.HTTP_500
            resp.text = dumps({"error": str(e)})
        finally:
            db.close()

    def insert_or_update(self, req, resp, id, query):
        """
        Insert or update an event.
        
        Args:
            req: The request object.
            resp: The response object.
            id: The event ID (optional).
            query: The SQL query to execute.
        """
        logger.info(f"Processing {'UPDATE' if id else 'INSERT'} request for event{f' {id}' if id else ''}")

        # Get request body payload (geojson Feature)
        try:
            body = req.stream.read().decode('utf-8')
            j = json.loads(body)
        except Exception as e:
            logger.error(f"Invalid JSON or bad encoding: {e}")
            resp.text = 'invalid json or bad encoding'
            resp.status = falcon.HTTP_400
            return

        resp.text = ''
        if "properties" not in j:
            resp.text = resp.text + "missing 'properties' elements\n"
            j['properties'] = dict()
        if "geometry" not in j:
            resp.text = resp.text + "missing 'geometry' elements\n"
            j['geometry'] = None
        if "when" not in j['properties'] and ("start" not in j['properties'] or "stop" not in j['properties']):
            resp.text = resp.text + "missing 'when' or 'start/stop' in properties\n"
            j['properties']['when'] = None
        if "type" not in j['properties']:
            resp.text = resp.text + "missing 'type' of event in properties\n"
            j['properties']['type'] = None
        if "what" not in j['properties']:
            resp.text = resp.text + "missing 'what' in properties\n"
            j['properties']['what'] = None
        if "type" in j and j['type'] != 'Feature':
            resp.text = resp.text + 'geojson must be "type":"Feature" only\n'
        if id is None and resp.text != '':
            resp.status = falcon.HTTP_400
            resp.set_header('Content-type', 'text/plain')
            return

        if 'when' in j['properties']:
            event_when = j['properties']['when']
        if "start" not in j['properties']:
            event_start = j['properties']['when']
        else:
            event_start = j['properties']['start']
        if "stop" not in j['properties']:
            event_stop = j['properties']['when']
        else:
            event_stop = j['properties']['stop']
        if event_start == event_stop:
            bounds = '[]'
        else:
            bounds = '[)'

        # Connect to db and insert
        db = db_connect()
        cur = db.cursor()

        # 'secret' based authentication
        if 'secret' in j['properties']:
            secret = cur.mogrify(" AND (events_tags->>'secret' = %s OR events_tags->>'secret' IS NULL) ", (j['properties']['secret'],)).decode("utf-8")
        elif 'secret' in req.params:
            secret = cur.mogrify(" AND (events_tags->>'secret' = %s OR events_tags->>'secret' IS NULL) ", (req.params['secret'],)).decode("utf-8")
        else:
            secret = " AND events_tags->>'secret' IS NULL "

        # Get the geometry part
        if j['geometry'] is not None:
            geometry = dumps(j['geometry'])
            h = self.maybe_insert_geometry(geometry, cur)
            if len(h) > 1 and h[1] is False:
                resp.text = "invalid geometry: %s\n" % h[2]
                resp.status = falcon.HTTP_400
                resp.set_header('Content-type', 'text/plain')
                return
        else:
            h = [None]
        params = (j['properties']['type'], j['properties']['what'], event_start, event_stop, bounds, dumps(j['properties']), h[0])
        if id:
            params = params + (id,)
        e = None
        rows = None
        try:
            sql = cur.mogrify(query, params)
            cur.execute(query.format(secret=secret), params)
            rows = cur.rowcount
            # Get newly created event id
            e = cur.fetchone()
            db.commit()
            logger.debug(f"SQL executed successfully: {sql}")
        except psycopg2.Error as err:
            logger.error(f"Database error: {err}, SQL: {sql}, Error: {err.pgerror}")
            db.rollback()
            pass

        # Send back to client
        if e is None:
            if id is None:
                cur.execute("""SELECT events_id FROM events WHERE events_what=%s
                        AND events_when=tstzrange(%s,%s,%s) AND events_geo=%s;""",
                        (j['properties']['what'], event_start, event_stop, bounds, h[0]))
            else:
                if rows == 0:
                    if 'secret' in req.params or 'secret' in j['properties']:
                        resp.status = '403 Unauthorized, secret does not match'
                    else:
                        resp.status = '403 Unauthorized, secret required'
                    return
                else:
                    cur.execute("""END; WITH s AS (SELECT * FROM events WHERE events_id = %s) SELECT e.events_id FROM events e, s WHERE e.events_what=coalesce(%s, s.events_what)
                        AND e.events_when=tstzrange(coalesce(%s, lower(s.events_when)),coalesce(%s,upper(s.events_when)),%s) AND e.events_geo=coalesce(%s, s.events_geo);""",
                        (id, j['properties']['what'], event_start, event_stop, bounds, h[0]))
            dupe = cur.fetchone()
            resp.text = """{"duplicate":"%s"}""" % (dupe[0])
            resp.status = '409 Conflict with event %s' % dupe[0]
            logger.warning(f"Duplicate event: {dupe[0]}")
        else:
            resp.text = """{"id":"%s"}""" % (e[0])
            if id is None:
                resp.status = falcon.HTTP_201
                logger.success(f"Event created with ID: {e[0]}")
            else:
                resp.status = falcon.HTTP_200
                logger.success(f"Event updated with ID: {e[0]}")

        cur.close()
        db.close()

    def on_post(self, req, resp):
        """
        Handle POST requests to the /event endpoint.
        Creates a new event.
        
        Args:
            req: The request object.
            resp: The response object.
        """
        logger.info("Processing POST request to /event")
        self.insert_or_update(req, resp, None, """INSERT INTO events ( events_type, events_what, events_when, events_tags, events_geo) VALUES (%s, %s, tstzrange(%s,%s,%s) , %s, %s) ON CONFLICT DO NOTHING RETURNING events_id;""")

    def on_put(self, req, resp, id):
        """
        Handle PUT requests to the /event/{id} endpoint.
        Updates an existing event (acts like PATCH).
        
        Args:
            req: The request object.
            resp: The response object.
            id: The event ID.
        """
        logger.info(f"Processing PUT request to /event/{id}")
        self.on_patch(req, resp, id)

    def on_patch(self, req, resp, id):
        """
        Handle PATCH requests to the /event/{id} endpoint.
        Updates an existing event.
        
        Args:
            req: The request object.
            resp: The response object.
            id: The event ID.
        """
        logger.info(f"Processing PATCH request to /event/{id}")
        # Coalesce are used to PATCH the data (new value may be NULL to keep the old one)
        self.insert_or_update(req, resp, id, """UPDATE events SET ( events_type, events_what, events_when, events_tags, events_geo) = (coalesce(%s, events_type), coalesce(%s, events_what), tstzrange(coalesce(%s, lower(events_when)),coalesce(%s, upper(events_when)),%s) , events_tags::jsonb || (%s::jsonb -'secret') , coalesce(%s, events_geo))
                              WHERE events_id = %s {secret} RETURNING events_id;""")

    def on_delete(self, req, resp, id):
        """
        Handle DELETE requests to the /event/{id} endpoint.
        Deletes an existing event.
        
        Args:
            req: The request object.
            resp: The response object.
            id: The event ID.
        """
        logger.info(f"Processing DELETE request to /event/{id}")
        
        db = db_connect()
        cur = db.cursor()
        
        try:
            cur.execute("""INSERT INTO events_deleted SELECT events_id, createdate, lastupdate, events_type, events_what, events_when, events_geo, events_tags FROM events WHERE events_id = %s """, (id,))
            rows_insert = cur.rowcount

            # 'secret' based authentication, must be null or same as during POST
            if 'secret' in req.params:
                cur.execute("""DELETE FROM events WHERE events_id = %s AND (events_tags->>'secret' = %s OR events_tags->>'secret' IS NULL)""", (id, req.params['secret']))
            else:
                cur.execute("""DELETE FROM events WHERE events_id = %s AND events_tags->>'secret' IS NULL;""", (id,))
            
            if cur.rowcount == 1:
                resp.status = "204 event deleted"
                db.commit()
                logger.success(f"Event deleted: {id}")
            elif rows_insert == 1:  # INSERT ok but DELETE fails due to missing secret...
                resp.status = "403 Unauthorized, secret needed to delete this event"
                db.rollback()
                logger.warning(f"Unauthorized deletion attempt for event: {id}")
            else:
                resp.status = "404 event not found"
                logger.warning(f"Event not found for deletion: {id}")
        except Exception as e:
            logger.error(f"Error deleting event {id}: {e}")
            resp.status = falcon.HTTP_500
            resp.text = dumps({"error": str(e)})
            db.rollback()
        finally:
            cur.close()
            db.close()

# Create a global instance of EventResource
event = EventResource()