Link-Log/backend/app/services/link_service.py

## Copyright © 2026 Olaf Kolkman
## SPDX-License-Identifier: GPL-3.0-or-later

from datetime import datetime, timezone
import json
import re
from urllib.parse import urlparse
from uuid import uuid4

from backend.app.core.security import clean_url
from backend.app.database import get_connection

MAX_TAGS = 10


def normalize_tags(tags: list[str] | None) -> list[str]:
    normalized = []
    normalized_keys = set()
    for tag in tags or []:
        value = tag.strip()
        if value and not value.startswith('#'):
            value = f'#{value}'
        key = value.casefold()
        if value and key not in normalized_keys:
            normalized.append(value)
            normalized_keys.add(key)
    if len(normalized) > MAX_TAGS:
        raise ValueError(f'A link can have at most {MAX_TAGS} tags')
    return normalized


def save_link_tags(conn, link_id: str, tags: list[str], user_id: str | None = None) -> list[str]:
    canonical_tags = []
    for tag in tags:
        tag_row = conn.execute(
            'SELECT id FROM tags WHERE lower(name) = lower(?)',
            (tag,),
        ).fetchone()
        if tag_row is None:
            conn.execute(
                'INSERT INTO tags (id, name, created_by) VALUES (?, ?, ?)',
                (str(uuid4()), tag, user_id),
            )
            tag_row = conn.execute('SELECT id FROM tags WHERE name = ?', (tag,)).fetchone()
        conn.execute(
            'INSERT OR IGNORE INTO link_tags (link_id, tag_id) VALUES (?, ?)',
            (link_id, tag_row['id']),
        )
        canonical_tags.append(conn.execute(
            'SELECT name FROM tags WHERE id = ?',
            (tag_row['id'],),
        ).fetchone()['name'])
    return canonical_tags


def get_link_tags(conn, link_id: str) -> list[str]:
    rows = conn.execute(
        '''
        SELECT tags.name FROM tags
        JOIN link_tags ON link_tags.tag_id = tags.id
        WHERE link_tags.link_id = ? ORDER BY tags.name
        ''',
        (link_id,),
    ).fetchall()
    return [row['name'] for row in rows]


def create_link(
    user_id: str,
    title: str,
    url: str,
    comment: str,
    timestamp: str | None,
    tags: list[str] | None = None,
):
    cleaned_url = clean_url(url)
    normalized_tags = normalize_tags(tags)
    created_at = datetime.now(timezone.utc).isoformat()
    record = {
        'id': str(uuid4()),
        'user_id': user_id,
        'title': title,
        'url': cleaned_url,
        'comment': comment,
        'timestamp': timestamp or created_at,
        'created_at': created_at,
        'updated_at': created_at,
        'is_public': 1,
    }
    with get_connection() as conn:
        conn.execute(
            '''
            INSERT INTO links (id, user_id, title, url, comment, timestamp, created_at, updated_at, is_public)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''',
            (
                record['id'],
                record['user_id'],
                record['title'],
                record['url'],
                record['comment'],
                record['timestamp'],
                record['created_at'],
                record['updated_at'],
                record['is_public'],
            ),
        )
        stored_tags = save_link_tags(conn, record['id'], normalized_tags, user_id=user_id)
        conn.commit()
    record['tags'] = stored_tags
    return record


def find_owned_link_by_title_url(user_id: str, title: str, url: str) -> dict | None:
    cleaned_url = clean_url(url)
    with get_connection() as conn:
        row = conn.execute(
            'SELECT * FROM links WHERE user_id = ? AND title = ? AND url = ? ORDER BY created_at DESC LIMIT 1',
            (user_id, title, cleaned_url),
        ).fetchone()
        if row is None:
            return None
        record = dict(row)
        record['tags'] = get_link_tags(conn, record['id'])
    return record


def find_owned_link_by_title(user_id: str, title: str) -> dict | None:
    with get_connection() as conn:
        row = conn.execute(
            'SELECT * FROM links WHERE user_id = ? AND title = ? ORDER BY created_at DESC LIMIT 1',
            (user_id, title),
        ).fetchone()
        if row is None:
            return None
        record = dict(row)
        record['tags'] = get_link_tags(conn, record['id'])
    return record


def _tokenize_search(value: str) -> list[str]:
    return [group1 or group2 for group1, group2 in re.findall(r'"([^"]+)"|(\S+)', value or '')]


def _parse_search_query(value: str) -> dict:
    groups = [[]]
    excluded = []
    sites = []
    for token in _tokenize_search(value):
        if token.upper() == 'OR':
            groups.append([])
            continue
        is_excluded = token.startswith('-') and len(token) > 1
        term = token[1:] if is_excluded else token
        site_match = re.match(r'^site:(\S+)$', term, re.IGNORECASE)
        if site_match and not is_excluded:
            sites.append(site_match.group(1).lower())
            continue
        if is_excluded:
            excluded.append(term.lower())
        else:
            groups[-1].append(term.lower())
    return {'groups': [group for group in groups if group], 'excluded': excluded, 'sites': sites}


def _searchable_text(record: dict) -> str:
    parts = [record.get('title'), record.get('url'), record.get('comment'), record.get('username'), *(record.get('tags') or [])]
    return ' '.join(part for part in parts if part).lower()


def _matches_search(record: dict, query: str) -> bool:
    if not query or not query.strip():
        return True
    parsed = _parse_search_query(query)
    text = _searchable_text(record)

    def site_matches(site: str) -> bool:
        hostname = (urlparse(record.get('url') or '').hostname or '').lower()
        return hostname == site or hostname.endswith(f'.{site}')

    matches_site = all(site_matches(site) for site in parsed['sites'])
    matches_group = not parsed['groups'] or any(all(term in text for term in group) for group in parsed['groups'])
    excludes_term = any(term in text for term in parsed['excluded'])
    return matches_site and matches_group and not excludes_term


def list_public_links(
    username: str | None = None,
    tag: str | None = None,
    search: str | None = None,
    sort: str = 'newest',
    page: int = 1,
    page_size: int | None = None,
) -> dict:
    order = 'ASC' if sort == 'oldest' else 'DESC'
    with get_connection() as conn:
        rows = conn.execute(
            f'''
            SELECT links.*, users.username, users.avatar_url, users.bio
            FROM links
            JOIN users ON users.id = links.user_id
            WHERE links.is_public = 1
              AND (? IS NULL OR users.username = ?)
              AND (
                ? IS NULL OR links.id IN (
                    SELECT link_tags.link_id FROM link_tags
                    JOIN tags ON tags.id = link_tags.tag_id
                    WHERE lower(tags.name) = lower(?)
                )
              )
            ORDER BY links.created_at {order}
            ''',
            (username, username, tag, tag),
        ).fetchall()
        records = [dict(row) for row in rows]
        for record in records:
            record['tags'] = get_link_tags(conn, record['id'])

    if search:
        records = [record for record in records if _matches_search(record, search)]

    total = len(records)
    if page_size:
        start = max(page - 1, 0) * page_size
        records = records[start:start + page_size]

    return {'items': records, 'total': total}


def get_public_profile(username: str) -> dict | None:
    with get_connection() as conn:
        row = conn.execute(
            'SELECT username, avatar_url, bio FROM users WHERE username = ?',
            (username,),
        ).fetchone()
    return dict(row) if row else None


def update_link(
    link_id: str,
    user_id: str,
    title: str,
    url: str,
    comment: str,
    tags: list[str] | None = None,
):
    cleaned_url = clean_url(url)
    normalized_tags = normalize_tags(tags)
    with get_connection() as conn:
        cursor = conn.execute(
            '''
            UPDATE links
            SET title = ?, url = ?, comment = ?, updated_at = CURRENT_TIMESTAMP
            WHERE id = ? AND user_id = ?
            ''',
            (title, cleaned_url, comment, link_id, user_id),
        )
        if cursor.rowcount == 0:
            return None
        conn.execute('DELETE FROM link_tags WHERE link_id = ?', (link_id,))
        stored_tags = save_link_tags(conn, link_id, normalized_tags, user_id=user_id)
        conn.commit()
        row = conn.execute('SELECT * FROM links WHERE id = ?', (link_id,)).fetchone()
        record = dict(row)
        record['tags'] = stored_tags
    return record


def delete_link(link_id: str, user_id: str) -> bool:
    with get_connection() as conn:
        cursor = conn.execute(
            'DELETE FROM links WHERE id = ? AND user_id = ?',
            (link_id, user_id),
        )
        conn.commit()
    return cursor.rowcount > 0


def get_owned_link(link_id: str, user_id: str) -> dict | None:
    with get_connection() as conn:
        row = conn.execute(
            'SELECT * FROM links WHERE id = ? AND user_id = ?',
            (link_id, user_id),
        ).fetchone()
    return dict(row) if row else None


def mark_mastodon_posted(link_id: str, user_id: str, post_id: str | None) -> bool:
    with get_connection() as conn:
        current = conn.execute(
            'SELECT mastodon_post_ids FROM links WHERE id = ? AND user_id = ?',
            (link_id, user_id),
        ).fetchone()
        post_ids = json.loads(current['mastodon_post_ids']) if current and current['mastodon_post_ids'] else []
        if post_id and post_id not in post_ids:
            post_ids.append(post_id)
        cursor = conn.execute(
            '''UPDATE links
               SET mastodon_posted = 1, mastodon_post_id = ?, mastodon_post_ids = ?, mastodon_posted_at = CURRENT_TIMESTAMP
               WHERE id = ? AND user_id = ?''',
            (post_id, json.dumps(post_ids), link_id, user_id),
        )
        conn.commit()
    return cursor.rowcount > 0


def list_public_users():
    with get_connection() as conn:
        rows = conn.execute(
            'SELECT username FROM users ORDER BY username'
        ).fetchall()
    return [row['username'] for row in rows]


def list_tags():
    with get_connection() as conn:
        rows = conn.execute('SELECT name FROM tags ORDER BY name').fetchall()
    tags = []
    seen = set()
    for row in rows:
        if row['name'].casefold() not in seen:
            tags.append(row['name'])
            seen.add(row['name'].casefold())
    return tags


def list_user_labels(user_id: str):
    with get_connection() as conn:
        rows = conn.execute(
            '''
            SELECT tags.id, tags.name, tags.created_by, users.username AS creator
            FROM tags
            LEFT JOIN users ON users.id = tags.created_by
            WHERE tags.created_by IS NULL
               OR tags.created_by = ?
               OR users.is_admin = 1
            ORDER BY tags.name
            ''',
            (user_id,),
        ).fetchall()
    return [dict(row) for row in rows]


def create_label(user_id: str, name: str):
    normalized = normalize_tags([name])
    label = normalized[0] if normalized else ''
    if not label:
        raise ValueError('Label cannot be empty')
    with get_connection() as conn:
        existing = conn.execute(
            'SELECT id FROM tags WHERE lower(name) = lower(?)', (label,)
        ).fetchone()
        if existing:
            raise ValueError('Label already exists')
        label_id = str(uuid4())
        conn.execute(
            'INSERT INTO tags (id, name, created_by) VALUES (?, ?, ?)',
            (label_id, label, user_id),
        )
        conn.commit()
    return {'id': label_id, 'name': label, 'created_by': user_id}


def update_label(label_id: str, user_id: str | None = None, name: str = '', is_admin: bool = False):
    normalized = normalize_tags([name])
    if not normalized:
        raise ValueError('Label cannot be empty')
    with get_connection() as conn:
        current = conn.execute(
            'SELECT id, name, created_by FROM tags WHERE id = ?', (label_id,)
        ).fetchone()
        if current is None or (not is_admin and current['created_by'] != user_id):
            return None
        duplicate = conn.execute(
            'SELECT id FROM tags WHERE lower(name) = lower(?) AND id != ?',
            (normalized[0], label_id),
        ).fetchone()
        if duplicate:
            raise ValueError('Label already exists')
        conn.execute('UPDATE tags SET name = ? WHERE id = ?', (normalized[0], label_id))
        conn.commit()
    return {'id': label_id, 'name': normalized[0], 'created_by': current['created_by']}


def delete_label(label_id: str, user_id: str | None = None, is_admin: bool = False):
    with get_connection() as conn:
        current = conn.execute(
            'SELECT id, created_by FROM tags WHERE id = ?', (label_id,)
        ).fetchone()
        if current is None or (not is_admin and current['created_by'] != user_id):
            return False
        conn.execute('DELETE FROM tags WHERE id = ?', (label_id,))
        conn.commit()
    return True