Update previous links

backend/app/api/links.py

@@ -2,11 +2,11 @@
 ## SPDX-License-Identifier: GPL-3.0-or-later
 
 import json
-from fastapi import APIRouter, Header, HTTPException, status
+from fastapi import APIRouter, Header, HTTPException, Response, status
 import logging
 from pydantic import BaseModel
 
-from backend.app.services.link_service import create_link, delete_link, get_link_tags, get_owned_link, list_public_links, list_tags, mark_mastodon_posted, update_link
+from backend.app.services.link_service import create_link, delete_link, find_owned_link_by_title_url, get_link_tags, get_owned_link, list_public_links, list_tags, mark_mastodon_posted, update_link
 from backend.app.database import get_connection
 from backend.app.services.plugin_manager import plugin_manager
 from backend.app.services.token_service import validate_token
@@ -35,8 +35,23 @@ def available_tags():
     return list_tags()
 
 
+@router.get('/links/check')
+def check_existing_link(
+    title: str,
+    url: str,
+    authorization: str | None = Header(default=None),
+):
+    if not authorization or not authorization.startswith('Bearer '):
+        raise HTTPException(status_code=401, detail='Missing or invalid Authorization header')
+    info = validate_token(authorization.replace('Bearer ', '', 1))
+    if info is None:
+        raise HTTPException(status_code=401, detail='Token expired or invalid')
+    record = find_owned_link_by_title_url(info['user_id'], title, url)
+    return {'exists': record is not None}
+
+
 @router.post('/links', status_code=status.HTTP_201_CREATED)
-def create_link_endpoint(payload: LinkCreate, authorization: str | None = Header(default=None)):
+def create_link_endpoint(payload: LinkCreate, response: Response, authorization: str | None = Header(default=None)):
     if not authorization or not authorization.startswith('Bearer '):
         raise HTTPException(status_code=401, detail='Missing or invalid Authorization header')
     token = authorization.replace('Bearer ', '', 1)
@@ -45,16 +60,23 @@ def create_link_endpoint(payload: LinkCreate, authorization: str | None = Header
         raise HTTPException(status_code=401, detail='Token expired or invalid')
 
     try:
-        record = create_link(info['user_id'], payload.title, payload.url, payload.comment, payload.timestamp, payload.tags)
+        record = find_owned_link_by_title_url(info['user_id'], payload.title, payload.url)
+        duplicate = record is not None
+        if duplicate:
+            record = update_link(record['id'], info['user_id'], payload.title, payload.url, payload.comment, payload.tags)
+        else:
+            record = create_link(info['user_id'], payload.title, payload.url, payload.comment, payload.timestamp, payload.tags)
     except ValueError as error:
         raise HTTPException(status_code=422, detail=str(error)) from error
+    if duplicate:
+        response.status_code = status.HTTP_200_OK
     plugin_results = plugin_manager.dispatch({'type': 'link_created', **record})
     mastodon_result = next((result for result in plugin_results if result.get('plugin') == 'mastodon'), None)
     if mastodon_result and mastodon_result.get('status') == 'posted':
         mark_mastodon_posted(record['id'], info['user_id'], mastodon_result.get('post_id'))
     if any(result.get('status') == 'failed' for result in plugin_results):
         logger.warning('One or more plugins failed for link_id=%s results=%s', record['id'], plugin_results)
-    return record
+    return {**record, 'duplicate': duplicate}
 
 
 @router.put('/links/{link_id}')

backend/app/services/link_service.py

@@ -109,6 +109,20 @@ def create_link(
     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 list_public_links(username: str | None = None):
     with get_connection() as conn:
         rows = conn.execute(

backend/tests/test_api.py

@@ -415,6 +415,45 @@ def test_submit_link_stores_cleaned_url_and_public_feed():
     assert 'alice' in users_response.json()
 
 
+def test_duplicate_link_updates_comment_tags_and_retriggers_plugins():
+    headers = login_headers()
+    payload = {
+        'title': 'Duplicate candidate',
+        'url': 'https://example.com/duplicate?utm_source=campaign',
+        'comment': 'first comment',
+        'tags': ['#First'],
+    }
+    first = client.post('/api/links', headers=headers, json=payload)
+    assert first.status_code == 201
+
+    with patch('backend.app.api.links.plugin_manager.dispatch', return_value=[]) as dispatch:
+        duplicate = client.post('/api/links', headers=headers, json={
+            **payload,
+            'comment': 'updated comment',
+            'tags': ['#Second'],
+        })
+    assert duplicate.status_code == 200
+    assert duplicate.json()['duplicate'] is True
+    assert duplicate.json()['id'] == first.json()['id']
+    dispatch.assert_called_once()
+    assert dispatch.call_args.args[0]['comment'] == 'updated comment'
+    assert dispatch.call_args.args[0]['tags'] == ['#Second']
+    feed_item = next(item for item in client.get('/api/public/feed').json() if item['id'] == first.json()['id'])
+    assert feed_item['comment'] == 'updated comment'
+    assert feed_item['tags'] == ['#Second']
+
+
+def test_duplicate_link_check_is_authenticated_and_detects_existing_entry():
+    headers = login_headers()
+    payload = {'title': 'Check candidate', 'url': 'https://example.com/check-candidate'}
+    assert client.get('/api/links/check', params=payload).status_code == 401
+    created = client.post('/api/links', headers=headers, json=payload)
+    assert created.status_code == 201
+    check = client.get('/api/links/check', headers=headers, params=payload)
+    assert check.status_code == 200
+    assert check.json()['exists'] is True
+
+
 def test_links_support_tags_and_tag_filtering():
     headers = login_headers()
     response = client.post('/api/links', headers=headers, json={