Merge pull request #575 from christianlouis/copilot/fix-imap-integration-ui-issues

feat(integrations): IMAP Gmail labels/delete controls, Watch Folder cloud sources, and comprehensive test coverage

app/models.py

@@ -544,9 +544,55 @@ class UserIntegration(Base):
     IMAP:
       config      = {"host": "imap.example.com", "port": 993,
                      "username": "user@example.com", "use_ssl": true,
-                     "delete_after_process": false}
+                     "delete_after_process": false,
+                     "gmail_apply_labels": true}
       credentials = {"password": "secret"}
 
+    WATCH_FOLDER (local):
+      config      = {"source_type": "local",
+                     "folder_path": "/data/inbox",
+                     "delete_after_process": false}
+
+    WATCH_FOLDER (s3):
+      config      = {"source_type": "s3", "bucket": "my-bucket",
+                     "region": "us-east-1", "prefix": "inbox/",
+                     "endpoint_url": null, "delete_after_process": false}
+      credentials = {"access_key_id": "AKI…", "secret_access_key": "…"}
+
+    WATCH_FOLDER (dropbox):
+      config      = {"source_type": "dropbox",
+                     "folder_path": "/Inbox/Scanner",
+                     "delete_after_process": false}
+      credentials = {"refresh_token": "…", "app_key": "…",
+                     "app_secret": "…"}
+
+    WATCH_FOLDER (google_drive):
+      config      = {"source_type": "google_drive",
+                     "folder_id": "1abc…",
+                     "delete_after_process": false}
+      credentials = {"credentials_json": "{…service-account…}"}
+
+    WATCH_FOLDER (onedrive):
+      config      = {"source_type": "onedrive",
+                     "folder_path": "/Documents/Inbox",
+                     "delete_after_process": false}
+      credentials = {"refresh_token": "…", "client_id": "…",
+                     "client_secret": "…"}
+
+    WATCH_FOLDER (nextcloud):
+      config      = {"source_type": "nextcloud",
+                     "url": "https://cloud.example.com",
+                     "folder_path": "/Documents/Inbox",
+                     "delete_after_process": false}
+      credentials = {"username": "user", "password": "secret"}
+
+    WATCH_FOLDER (webdav):
+      config      = {"source_type": "webdav",
+                     "url": "https://webdav.example.com/dav/",
+                     "folder_path": "/remote.php/webdav/Inbox",
+                     "delete_after_process": false}
+      credentials = {"username": "user", "password": "secret"}
+
     S3:
       config      = {"bucket": "my-bucket", "region": "us-east-1",
                      "endpoint_url": null, "folder_prefix": ""}

app/tasks/imap_tasks.py

@@ -249,6 +249,7 @@ def _pull_user_integration_imap() -> None:
                     password = creds.get("password")
                     use_ssl = cfg.get("use_ssl", True)
                     delete_after = cfg.get("delete_after_process", False)
+                    gmail_labels = cfg.get("gmail_apply_labels", True)
 
                     if not (host and username and password):
                         logger.warning(
@@ -267,6 +268,7 @@ def _pull_user_integration_imap() -> None:
                         use_ssl=use_ssl,
                         delete_after_process=delete_after,
                         owner_id=integ.owner_id,
+                        gmail_apply_labels=gmail_labels,
                     )
                     integ.last_used_at = datetime.now(timezone.utc)
                     integ.last_error = None
@@ -317,7 +319,17 @@ def check_and_pull_mailbox(
     )
 
 
-def pull_inbox(mailbox_key, host, port, username, password, use_ssl, delete_after_process, owner_id=None):
+def pull_inbox(
+    mailbox_key,
+    host,
+    port,
+    username,
+    password,
+    use_ssl,
+    delete_after_process,
+    owner_id=None,
+    gmail_apply_labels=True,
+):
     """
     Connects to the IMAP inbox, fetches new unread emails from the last 3 days,
     and processes attachments while preserving the original unread status.
@@ -331,6 +343,8 @@ def pull_inbox(mailbox_key, host, port, username, password, use_ssl, delete_afte
     Args:
         owner_id: Optional user identifier. When provided, ingested documents are
                   attributed to this user via ``process_document`` / ``convert_to_pdf``.
+        gmail_apply_labels: Whether to apply Gmail-specific labels and stars to
+                  processed emails. Only relevant for Gmail hosts. Defaults to True.
     """
     logger.info("Connecting to %s at %s:%s (SSL=%s)", mailbox_key, host, port, use_ssl)
     processed_emails = load_processed_emails()
@@ -386,7 +400,7 @@ def pull_inbox(mailbox_key, host, port, username, password, use_ssl, delete_afte
                 continue
 
             # For Gmail, check if the email already has the "Ingested" label.
-            if is_gmail_host:
+            if is_gmail_host and gmail_apply_labels:
                 if email_already_has_label(mail, num, "Ingested"):
                     logger.info("Skipping email %s in %s, already labeled 'Ingested'.", msg_id, mailbox_key)
                     continue
@@ -398,7 +412,7 @@ def pull_inbox(mailbox_key, host, port, username, password, use_ssl, delete_afte
             if settings.imap_readonly_mode:
                 logger.info("Readonly mode: skipping mailbox modifications for %s in %s", msg_id, mailbox_key)
             else:
-                if is_gmail_host:
+                if is_gmail_host and gmail_apply_labels:
                     mark_as_processed_with_star(mail, num)
                     mark_as_processed_with_label(mail, num, label="Ingested")
 

app/tasks/watch_folder_tasks.py

@@ -16,6 +16,7 @@ import json
 import logging
 import os
 from datetime import datetime, timedelta, timezone
+from typing import Any
 
 import redis
 from celery import shared_task
@@ -1424,16 +1425,724 @@ def _scan_user_watch_folder(
     return count
 
 
+# ---------------------------------------------------------------------------
+# Per-user cloud source watch folder scanning
+# ---------------------------------------------------------------------------
+
+# Maps source_type values to their per-user scan functions.
+_USER_WF_CLOUD_HANDLERS: dict[str, Any] = {}  # populated after function defs
+
+
+def _scan_user_s3_folder(
+    cfg: dict,
+    creds: dict,
+    cache: dict[str, str],
+    delete_after: bool,
+    owner_id: str,
+) -> int:
+    """Scan an S3 bucket prefix using per-user credentials.
+
+    Args:
+        cfg: Integration config with ``bucket``, ``region``, ``prefix``, ``endpoint_url``.
+        creds: Decrypted credentials with ``access_key_id``, ``secret_access_key``.
+        cache: In-memory dict of already-processed file keys.
+        delete_after: Whether to remove the source object after ingestion.
+        owner_id: The user to attribute ingested documents to.
+
+    Returns:
+        Number of files newly enqueued.
+    """
+    try:
+        import boto3
+        from botocore.exceptions import ClientError
+    except ImportError as exc:
+        logger.error("User S3 watch folder: boto3 not installed: %s", exc)
+        return 0
+
+    bucket = cfg.get("bucket", "")
+    prefix = cfg.get("prefix", "")
+    region = cfg.get("region", "us-east-1")
+    endpoint_url = cfg.get("endpoint_url") or None
+
+    if not bucket:
+        logger.warning("User S3 watch folder: bucket not configured.")
+        return 0
+
+    access_key = creds.get("access_key_id", "")
+    secret_key = creds.get("secret_access_key", "")
+    if not (access_key and secret_key):
+        logger.warning("User S3 watch folder: credentials incomplete.")
+        return 0
+
+    try:
+        client_kwargs: dict = {
+            "region_name": region,
+            "aws_access_key_id": access_key,
+            "aws_secret_access_key": secret_key,
+        }
+        if endpoint_url:
+            client_kwargs["endpoint_url"] = endpoint_url
+        s3 = boto3.client("s3", **client_kwargs)
+    except Exception as exc:
+        logger.error("User S3 watch folder: failed to create client: %s", exc)
+        return 0
+
+    count = 0
+    paginator = s3.get_paginator("list_objects_v2")
+
+    try:
+        pages = paginator.paginate(Bucket=bucket, Prefix=prefix)
+    except Exception as exc:
+        logger.error("User S3 watch folder: failed to list %s/%s: %s", bucket, prefix, exc)
+        return 0
+
+    for page in pages:
+        for obj in page.get("Contents", []):
+            key = obj["Key"]
+            filename = key.split("/")[-1]
+            if not filename or not _is_allowed_file(filename):
+                continue
+
+            cache_key = f"s3:{bucket}/{key}"
+            if cache_key in cache:
+                continue
+
+            dest_path = os.path.join(settings.workdir, f"uwf_s3_{owner_id}_{filename}")
+            if os.path.exists(dest_path):
+                base2, ext2 = os.path.splitext(f"uwf_s3_{owner_id}_{filename}")
+                dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}")
+
+            try:
+                s3.download_file(bucket, key, dest_path)
+                logger.info("User S3 watch folder: downloaded s3://%s/%s", bucket, key)
+            except ClientError as exc:
+                logger.error("User S3 watch folder: failed to download %s: %s", key, exc)
+                if os.path.exists(dest_path):
+                    os.remove(dest_path)
+                continue
+
+            _enqueue_file(dest_path, filename=filename, owner_id=owner_id)
+            _mark_processed(cache, cache_key)
+            count += 1
+
+            if delete_after:
+                try:
+                    s3.delete_object(Bucket=bucket, Key=key)
+                    logger.info("User S3 watch folder: deleted s3://%s/%s", bucket, key)
+                except Exception as exc:
+                    logger.warning("User S3 watch folder: could not delete %s: %s", key, exc)
+
+    return count
+
+
+def _scan_user_dropbox_folder(
+    cfg: dict,
+    creds: dict,
+    cache: dict[str, str],
+    delete_after: bool,
+    owner_id: str,
+) -> int:
+    """Scan a Dropbox folder using per-user credentials.
+
+    Args:
+        cfg: Integration config with ``folder_path``.
+        creds: Decrypted credentials with ``refresh_token``, ``app_key``, ``app_secret``.
+        cache: In-memory dict of already-processed file keys.
+        delete_after: Whether to remove the source file after ingestion.
+        owner_id: The user to attribute ingested documents to.
+
+    Returns:
+        Number of files newly enqueued.
+    """
+    try:
+        import dropbox as dropbox_module
+    except ImportError as exc:
+        logger.error("User Dropbox watch folder: dropbox SDK not installed: %s", exc)
+        return 0
+
+    refresh_token = creds.get("refresh_token", "")
+    app_key = creds.get("app_key", "")
+    app_secret = creds.get("app_secret", "")
+    if not (refresh_token and app_key and app_secret):
+        logger.warning("User Dropbox watch folder: credentials incomplete.")
+        return 0
+
+    try:
+        dbx = dropbox_module.Dropbox(
+            oauth2_refresh_token=refresh_token,
+            app_key=app_key,
+            app_secret=app_secret,
+        )
+    except Exception as exc:
+        logger.error("User Dropbox watch folder: auth failed: %s", exc)
+        return 0
+
+    folder_path = cfg.get("folder_path", "")
+    if not folder_path:
+        logger.warning("User Dropbox watch folder: folder_path not configured.")
+        return 0
+
+    count = 0
+    try:
+        result = dbx.files_list_folder(folder_path)
+        entries = list(result.entries)
+        while result.has_more:
+            result = dbx.files_list_folder_continue(result.cursor)
+            entries.extend(result.entries)
+    except Exception as exc:
+        logger.error("User Dropbox watch folder: cannot list %s: %s", folder_path, exc)
+        return 0
+
+    for entry in entries:
+        if not isinstance(entry, dropbox_module.files.FileMetadata):
+            continue
+
+        filename = entry.name
+        if not _is_allowed_file(filename):
+            continue
+
+        cache_key = f"dropbox:{entry.id}"
+        if cache_key in cache:
+            continue
+
+        dest_path = os.path.join(settings.workdir, f"uwf_dbx_{owner_id}_{filename}")
+        if os.path.exists(dest_path):
+            base2, ext2 = os.path.splitext(f"uwf_dbx_{owner_id}_{filename}")
+            dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}")
+
+        try:
+            _meta, response = dbx.files_download(entry.path_lower)
+            with open(dest_path, "wb") as f:
+                f.write(response.content)
+            logger.info("User Dropbox watch folder: downloaded %s", filename)
+        except Exception as exc:
+            logger.error("User Dropbox watch folder: failed to download %s: %s", filename, exc)
+            if os.path.exists(dest_path):
+                os.remove(dest_path)
+            continue
+
+        _enqueue_file(dest_path, filename=filename, owner_id=owner_id)
+        _mark_processed(cache, cache_key)
+        count += 1
+
+        if delete_after:
+            try:
+                dbx.files_delete_v2(entry.path_lower)
+                logger.info("User Dropbox watch folder: deleted %s", entry.path_lower)
+            except Exception as exc:
+                logger.warning("User Dropbox watch folder: could not delete %s: %s", entry.path_lower, exc)
+
+    return count
+
+
+def _scan_user_google_drive_folder(
+    cfg: dict,
+    creds: dict,
+    cache: dict[str, str],
+    delete_after: bool,
+    owner_id: str,
+) -> int:
+    """Scan a Google Drive folder using per-user service-account credentials.
+
+    Args:
+        cfg: Integration config with ``folder_id``.
+        creds: Decrypted credentials with ``credentials_json``.
+        cache: In-memory dict of already-processed file keys.
+        delete_after: Whether to remove the source file after ingestion.
+        owner_id: The user to attribute ingested documents to.
+
+    Returns:
+        Number of files newly enqueued.
+    """
+    try:
+        from google.oauth2 import service_account
+        from googleapiclient.discovery import build
+    except ImportError as exc:
+        logger.error("User Google Drive watch folder: SDK not installed: %s", exc)
+        return 0
+
+    creds_json = creds.get("credentials_json", "")
+    if not creds_json:
+        logger.warning("User Google Drive watch folder: credentials_json not provided.")
+        return 0
+
+    folder_id = cfg.get("folder_id", "")
+    if not folder_id:
+        logger.warning("User Google Drive watch folder: folder_id not configured.")
+        return 0
+
+    try:
+        import json as _json
+
+        info = _json.loads(creds_json) if isinstance(creds_json, str) else creds_json
+        credentials = service_account.Credentials.from_service_account_info(
+            info,
+            scopes=["https://www.googleapis.com/auth/drive"],
+        )
+        service = build("drive", "v3", credentials=credentials)
+    except Exception as exc:
+        logger.error("User Google Drive watch folder: auth failed: %s", exc)
+        return 0
+
+    count = 0
+    query = f"'{folder_id}' in parents and trashed = false and mimeType != 'application/vnd.google-apps.folder'"
+    page_token = None
+
+    while True:
+        try:
+            params: dict = {
+                "q": query,
+                "fields": "nextPageToken, files(id, name, mimeType)",
+                "pageSize": 100,
+            }
+            if page_token:
+                params["pageToken"] = page_token
+            response = service.files().list(**params).execute()
+        except Exception as exc:
+            logger.error("User Google Drive watch folder: listing %s failed: %s", folder_id, exc)
+            break
+
+        for file_meta in response.get("files", []):
+            file_id_gd = file_meta["id"]
+            filename = file_meta["name"]
+
+            if not _is_allowed_file(filename):
+                continue
+
+            cache_key = f"gdrive:{file_id_gd}"
+            if cache_key in cache:
+                continue
+
+            dest_path = os.path.join(settings.workdir, f"uwf_gd_{owner_id}_{filename}")
+            if os.path.exists(dest_path):
+                base2, ext2 = os.path.splitext(f"uwf_gd_{owner_id}_{filename}")
+                dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}")
+
+            try:
+                import io
+
+                from googleapiclient.http import MediaIoBaseDownload
+
+                request = service.files().get_media(fileId=file_id_gd)
+                buf = io.BytesIO()
+                downloader = MediaIoBaseDownload(buf, request)
+                done = False
+                while not done:
+                    _, done = downloader.next_chunk()
+                with open(dest_path, "wb") as f:
+                    f.write(buf.getvalue())
+                logger.info("User Google Drive watch folder: downloaded %s", filename)
+            except Exception as exc:
+                logger.error("User Google Drive watch folder: download %s failed: %s", filename, exc)
+                if os.path.exists(dest_path):
+                    os.remove(dest_path)
+                continue
+
+            _enqueue_file(dest_path, filename=filename, owner_id=owner_id)
+            _mark_processed(cache, cache_key)
+            count += 1
+
+            if delete_after:
+                try:
+                    service.files().delete(fileId=file_id_gd).execute()
+                    logger.info("User Google Drive watch folder: deleted %s", filename)
+                except Exception as exc:
+                    logger.warning("User Google Drive watch folder: could not delete %s: %s", filename, exc)
+
+        page_token = response.get("nextPageToken")
+        if not page_token:
+            break
+
+    return count
+
+
+def _scan_user_onedrive_folder(
+    cfg: dict,
+    creds: dict,
+    cache: dict[str, str],
+    delete_after: bool,
+    owner_id: str,
+) -> int:
+    """Scan a OneDrive folder using per-user OAuth credentials.
+
+    Args:
+        cfg: Integration config with ``folder_path``.
+        creds: Decrypted credentials with ``refresh_token``, ``client_id``, ``client_secret``.
+        cache: In-memory dict of already-processed file keys.
+        delete_after: Whether to remove the source file after ingestion.
+        owner_id: The user to attribute ingested documents to.
+
+    Returns:
+        Number of files newly enqueued.
+    """
+    import requests as req_lib
+
+    refresh_token = creds.get("refresh_token", "")
+    client_id = creds.get("client_id", "")
+    client_secret = creds.get("client_secret", "")
+    if not (refresh_token and client_id and client_secret):
+        logger.warning("User OneDrive watch folder: credentials incomplete.")
+        return 0
+
+    folder_path = cfg.get("folder_path", "")
+    if not folder_path:
+        logger.warning("User OneDrive watch folder: folder_path not configured.")
+        return 0
+
+    # Exchange refresh token for an access token
+    try:
+        token_resp = req_lib.post(
+            "https://login.microsoftonline.com/common/oauth2/v2.0/token",
+            data={
+                "grant_type": "refresh_token",
+                "refresh_token": refresh_token,
+                "client_id": client_id,
+                "client_secret": client_secret,
+                "scope": "https://graph.microsoft.com/.default",
+            },
+            timeout=getattr(settings, "http_request_timeout", 120),
+        )
+        token_resp.raise_for_status()
+        access_token = token_resp.json()["access_token"]
+    except Exception as exc:
+        logger.error("User OneDrive watch folder: token exchange failed: %s", exc)
+        return 0
+
+    headers = {"Authorization": f"Bearer {access_token}"}
+    import urllib.parse
+
+    encoded_path = urllib.parse.quote(folder_path.lstrip("/"))
+    list_url: str | None = f"https://graph.microsoft.com/v1.0/me/drive/root:/{encoded_path}:/children"
+
+    count = 0
+    timeout = getattr(settings, "http_request_timeout", 120)
+
+    while list_url:
+        try:
+            resp = req_lib.get(list_url, headers=headers, timeout=timeout)
+            resp.raise_for_status()
+            data = resp.json()
+        except Exception as exc:
+            logger.error("User OneDrive watch folder: listing %s failed: %s", folder_path, exc)
+            break
+
+        for item in data.get("value", []):
+            if "folder" in item:
+                continue
+
+            filename = item["name"]
+            item_id = item["id"]
+
+            if not _is_allowed_file(filename):
+                continue
+
+            cache_key = f"onedrive:{item_id}"
+            if cache_key in cache:
+                continue
+
+            download_url = item.get("@microsoft.graph.downloadUrl")
+            if not download_url:
+                continue
+
+            dest_path = os.path.join(settings.workdir, f"uwf_od_{owner_id}_{filename}")
+            if os.path.exists(dest_path):
+                base2, ext2 = os.path.splitext(f"uwf_od_{owner_id}_{filename}")
+                dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}")
+
+            try:
+                dl_resp = req_lib.get(download_url, headers=headers, timeout=timeout)
+                dl_resp.raise_for_status()
+                with open(dest_path, "wb") as f:
+                    f.write(dl_resp.content)
+                logger.info("User OneDrive watch folder: downloaded %s", filename)
+            except Exception as exc:
+                logger.error("User OneDrive watch folder: download %s failed: %s", filename, exc)
+                if os.path.exists(dest_path):
+                    os.remove(dest_path)
+                continue
+
+            _enqueue_file(dest_path, filename=filename, owner_id=owner_id)
+            _mark_processed(cache, cache_key)
+            count += 1
+
+            if delete_after:
+                try:
+                    del_resp = req_lib.delete(
+                        f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}",
+                        headers=headers,
+                        timeout=timeout,
+                    )
+                    del_resp.raise_for_status()
+                    logger.info("User OneDrive watch folder: deleted %s", filename)
+                except Exception as exc:
+                    logger.warning("User OneDrive watch folder: could not delete %s: %s", filename, exc)
+
+        list_url = data.get("@odata.nextLink")
+
+    return count
+
+
+def _scan_user_nextcloud_folder(
+    cfg: dict,
+    creds: dict,
+    cache: dict[str, str],
+    delete_after: bool,
+    owner_id: str,
+) -> int:
+    """Scan a Nextcloud folder using per-user WebDAV credentials.
+
+    Args:
+        cfg: Integration config with ``url``, ``folder_path``.
+        creds: Decrypted credentials with ``username``, ``password``.
+        cache: In-memory dict of already-processed file keys.
+        delete_after: Whether to remove the source file after ingestion.
+        owner_id: The user to attribute ingested documents to.
+
+    Returns:
+        Number of files newly enqueued.
+    """
+    import defusedxml.ElementTree as ET
+    import requests as req_lib
+    from requests.auth import HTTPBasicAuth
+
+    nc_url = cfg.get("url", "")
+    folder_path = cfg.get("folder_path", "")
+    nc_user = creds.get("username", "")
+    nc_pass = creds.get("password", "")
+
+    if not (nc_url and nc_user and nc_pass):
+        logger.warning("User Nextcloud watch folder: connection settings incomplete.")
+        return 0
+
+    auth = HTTPBasicAuth(nc_user, nc_pass)
+    timeout = getattr(settings, "http_request_timeout", 120)
+
+    base = nc_url.rstrip("/")
+    folder = folder_path.strip("/")
+    propfind_url = f"{base}/{folder}/" if folder else f"{base}/"
+
+    try:
+        resp = req_lib.request(
+            "PROPFIND",
+            propfind_url,
+            auth=auth,
+            headers={"Depth": "1", "Content-Type": "application/xml"},
+            timeout=timeout,
+        )
+        resp.raise_for_status()
+    except Exception as exc:
+        logger.error("User Nextcloud watch folder: PROPFIND on %s failed: %s", propfind_url, exc)
+        return 0
+
+    count = 0
+    try:
+        root = ET.fromstring(resp.text)  # noqa: S314 — defusedxml is safe
+    except Exception as exc:
+        logger.error("User Nextcloud watch folder: failed to parse response: %s", exc)
+        return 0
+
+    ns = {"d": "DAV:"}
+    for response_el in root.findall("d:response", ns):
+        href_el = response_el.find("d:href", ns)
+        if href_el is None or href_el.text is None:
+            continue
+
+        href = href_el.text
+        if href.rstrip("/").endswith(folder.rstrip("/")):
+            continue
+
+        import urllib.parse
+
+        filename = urllib.parse.unquote(href.rstrip("/").split("/")[-1])
+        if not _is_allowed_file(filename):
+            continue
+
+        cache_key = f"nextcloud:{href}"
+        if cache_key in cache:
+            continue
+
+        if href.startswith("http"):
+            file_url = href
+        else:
+            from urllib.parse import urlparse
+
+            parsed = urlparse(nc_url)
+            file_url = f"{parsed.scheme}://{parsed.netloc}{href}"
+
+        dest_path = os.path.join(settings.workdir, f"uwf_nc_{owner_id}_{filename}")
+        if os.path.exists(dest_path):
+            base_name, ext2 = os.path.splitext(f"uwf_nc_{owner_id}_{filename}")
+            dest_path = os.path.join(settings.workdir, f"{base_name}_{int(datetime.now().timestamp())}{ext2}")
+
+        try:
+            dl = req_lib.get(file_url, auth=auth, timeout=timeout)
+            dl.raise_for_status()
+            with open(dest_path, "wb") as f:
+                f.write(dl.content)
+            logger.info("User Nextcloud watch folder: downloaded %s", filename)
+        except Exception as exc:
+            logger.error("User Nextcloud watch folder: download %s failed: %s", filename, exc)
+            if os.path.exists(dest_path):
+                os.remove(dest_path)
+            continue
+
+        _enqueue_file(dest_path, filename=filename, owner_id=owner_id)
+        _mark_processed(cache, cache_key)
+        count += 1
+
+        if delete_after:
+            try:
+                del_resp = req_lib.request("DELETE", file_url, auth=auth, timeout=timeout)
+                del_resp.raise_for_status()
+                logger.info("User Nextcloud watch folder: deleted %s", filename)
+            except Exception as exc:
+                logger.warning("User Nextcloud watch folder: could not delete %s: %s", filename, exc)
+
+    return count
+
+
+def _scan_user_webdav_folder(
+    cfg: dict,
+    creds: dict,
+    cache: dict[str, str],
+    delete_after: bool,
+    owner_id: str,
+) -> int:
+    """Scan a WebDAV folder using per-user credentials.
+
+    Args:
+        cfg: Integration config with ``url``, ``folder_path``.
+        creds: Decrypted credentials with ``username``, ``password``.
+        cache: In-memory dict of already-processed file keys.
+        delete_after: Whether to remove the source file after ingestion.
+        owner_id: The user to attribute ingested documents to.
+
+    Returns:
+        Number of files newly enqueued.
+    """
+    import defusedxml.ElementTree as ET
+    import requests as req_lib
+    from requests.auth import HTTPBasicAuth
+
+    webdav_url = cfg.get("url", "")
+    folder_path = cfg.get("folder_path", "")
+    dav_user = creds.get("username", "")
+    dav_pass = creds.get("password", "")
+
+    if not webdav_url:
+        logger.warning("User WebDAV watch folder: URL not configured.")
+        return 0
+
+    base = webdav_url.rstrip("/")
+    folder = folder_path.strip("/")
+    propfind_url = f"{base}/{folder}/" if folder else f"{base}/"
+
+    auth = HTTPBasicAuth(dav_user, dav_pass) if dav_user else None
+    timeout = getattr(settings, "http_request_timeout", 120)
+
+    try:
+        resp = req_lib.request(
+            "PROPFIND",
+            propfind_url,
+            auth=auth,
+            headers={"Depth": "1"},
+            timeout=timeout,
+        )
+        resp.raise_for_status()
+    except Exception as exc:
+        logger.error("User WebDAV watch folder: PROPFIND on %s failed: %s", propfind_url, exc)
+        return 0
+
+    count = 0
+    try:
+        root = ET.fromstring(resp.text)  # noqa: S314 — defusedxml is safe
+    except Exception as exc:
+        logger.error("User WebDAV watch folder: failed to parse response: %s", exc)
+        return 0
+
+    from urllib.parse import unquote, urlparse
+
+    ns = {"d": "DAV:"}
+    for response_el in root.findall("d:response", ns):
+        href_el = response_el.find("d:href", ns)
+        if href_el is None or href_el.text is None:
+            continue
+
+        href = href_el.text
+        if href.endswith("/"):
+            continue
+
+        filename = unquote(href.split("/")[-1])
+        if not _is_allowed_file(filename):
+            continue
+
+        cache_key = f"webdav:{href}"
+        if cache_key in cache:
+            continue
+
+        if href.startswith("http"):
+            file_url = href
+        else:
+            parsed = urlparse(webdav_url)
+            file_url = f"{parsed.scheme}://{parsed.netloc}{href}"
+
+        dest_path = os.path.join(settings.workdir, f"uwf_dav_{owner_id}_{filename}")
+        if os.path.exists(dest_path):
+            base2, ext2 = os.path.splitext(f"uwf_dav_{owner_id}_{filename}")
+            dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}")
+
+        try:
+            dl = req_lib.get(file_url, auth=auth, timeout=timeout)
+            dl.raise_for_status()
+            with open(dest_path, "wb") as f:
+                f.write(dl.content)
+            logger.info("User WebDAV watch folder: downloaded %s", filename)
+        except Exception as exc:
+            logger.error("User WebDAV watch folder: download %s failed: %s", filename, exc)
+            if os.path.exists(dest_path):
+                os.remove(dest_path)
+            continue
+
+        _enqueue_file(dest_path, filename=filename, owner_id=owner_id)
+        _mark_processed(cache, cache_key)
+        count += 1
+
+        if delete_after:
+            try:
+                del_resp = req_lib.request("DELETE", file_url, auth=auth, timeout=timeout)
+                del_resp.raise_for_status()
+                logger.info("User WebDAV watch folder: deleted %s", filename)
+            except Exception as exc:
+                logger.warning("User WebDAV watch folder: could not delete %s: %s", filename, exc)
+
+    return count
+
+
+# Populate the cloud handler dispatch table
+_USER_WF_CLOUD_HANDLERS.update(
+    {
+        "s3": _scan_user_s3_folder,
+        "dropbox": _scan_user_dropbox_folder,
+        "google_drive": _scan_user_google_drive_folder,
+        "onedrive": _scan_user_onedrive_folder,
+        "nextcloud": _scan_user_nextcloud_folder,
+        "webdav": _scan_user_webdav_folder,
+    }
+)
+
+
 def _pull_user_integration_watch_folders() -> dict:
-    """Iterate over all active WATCH_FOLDER UserIntegrations and scan their paths.
+    """Iterate over all active WATCH_FOLDER UserIntegrations and scan their sources.
 
     Polls the ``user_integrations`` table for records with
     ``integration_type='WATCH_FOLDER'``, ``direction='SOURCE'``, and
     ``is_active=True``.  Each integration's config is decoded and the
-    configured ``folder_path`` is scanned for new files, which are enqueued
+    configured source is scanned for new files, which are enqueued
     with the owning user's ``owner_id``.
 
-    Path traversal protection is enforced on the configured path.
+    Path traversal protection is enforced on local filesystem paths.
+    Cloud source types (S3, Dropbox, Google Drive, OneDrive, Nextcloud, WebDAV)
+    are dispatched to their per-user scanning helpers.
 
     Individual integration failures are caught and recorded without crashing
     the polling loop.
@@ -1445,6 +2154,7 @@ def _pull_user_integration_watch_folders() -> dict:
         import json as _json
 
         from app.models import IntegrationDirection, IntegrationType, UserIntegration
+        from app.utils.encryption import decrypt_value
 
         db = _get_db_session()
         try:
@@ -1462,32 +2172,51 @@ def _pull_user_integration_watch_folders() -> dict:
             for integ in integrations:
                 try:
                     cfg = _json.loads(integ.config) if integ.config else {}
-                    folder_path = cfg.get("folder_path", "")
                     delete_after = cfg.get("delete_after_process", False)
-
+                    source_type = cfg.get("source_type", "local")
-                    if not folder_path:
-                        logger.warning(
-                            "Watch folder integration %d (owner %s) has no folder_path — skipping.",
-                            integ.id,
-                            integ.owner_id,
-                        )
-                        continue
-
-                    if not _is_safe_watch_path(folder_path):
-                        error_msg = f"Unsafe watch folder path rejected: {folder_path}"
-                        logger.error(
-                            "Watch folder integration %d (owner %s): %s",
-                            integ.id,
-                            integ.owner_id,
-                            error_msg,
-                        )
-                        integ.last_error = error_msg[:_MAX_ERROR_LENGTH]
-                        db.commit()
-                        continue
 
                     cache_file = f"{_USER_WF_CACHE_PREFIX}{integ.id}.json"
                     cache = _load_cache(cache_file)
-                    n = _scan_user_watch_folder(folder_path, cache, delete_after, integ.owner_id)
+
+                    if source_type == "local":
+                        # Local filesystem watch folder (original behaviour)
+                        folder_path = cfg.get("folder_path", "")
+                        if not folder_path:
+                            logger.warning(
+                                "Watch folder integration %d (owner %s) has no folder_path — skipping.",
+                                integ.id,
+                                integ.owner_id,
+                            )
+                            continue
+
+                        if not _is_safe_watch_path(folder_path):
+                            error_msg = f"Unsafe watch folder path rejected: {folder_path}"
+                            logger.error(
+                                "Watch folder integration %d (owner %s): %s",
+                                integ.id,
+                                integ.owner_id,
+                                error_msg,
+                            )
+                            integ.last_error = error_msg[:_MAX_ERROR_LENGTH]
+                            db.commit()
+                            continue
+
+                        n = _scan_user_watch_folder(folder_path, cache, delete_after, integ.owner_id)
+                    elif source_type in _USER_WF_CLOUD_HANDLERS:
+                        # Cloud source — decrypt per-user credentials and delegate
+                        raw_creds = decrypt_value(integ.credentials) if integ.credentials else None
+                        creds = _json.loads(raw_creds) if raw_creds else {}
+                        handler = _USER_WF_CLOUD_HANDLERS[source_type]
+                        n = handler(cfg, creds, cache, delete_after, integ.owner_id)
+                    else:
+                        logger.warning(
+                            "Watch folder integration %d (owner %s): unknown source_type '%s' — skipping.",
+                            integ.id,
+                            integ.owner_id,
+                            source_type,
+                        )
+                        continue
+
                     _save_cache(cache_file, cache)
 
                     total_files += n
@@ -1495,11 +2224,11 @@ def _pull_user_integration_watch_folders() -> dict:
                     integ.last_error = None
                     db.commit()
                     logger.info(
-                        "Watch folder integration %d (owner %s): %d file(s) enqueued from %s",
+                        "Watch folder integration %d (owner %s): %d file(s) enqueued (source=%s)",
                         integ.id,
                         integ.owner_id,
                         n,
-                        folder_path,
+                        source_type,
                     )
                 except Exception as exc:  # noqa: BLE001
                     error_msg = str(exc)[:_MAX_ERROR_LENGTH]

docs/ConfigurationGuide.md

@@ -272,10 +272,16 @@ DocuElevate can poll a WebDAV folder for new files. It reuses the existing WebDA
 In addition to system-level watch folders, each user can configure personal watch folder sources through the **Integrations** dashboard (`/integrations`). Documents ingested from per-user watch folder integrations are automatically attributed to the owning user's `owner_id`.
 
 Per-user watch folder integrations are stored in the `user_integrations` table with `integration_type='WATCH_FOLDER'` and `direction='SOURCE'`. The `config` JSON field stores:
-- `folder_path` — absolute path to the directory to scan
+- `source_type` — the type of source to scan (`local`, `s3`, `dropbox`, `google_drive`, `onedrive`, `nextcloud`, `webdav`; default: `local`)
+- `folder_path` — path to the directory/folder to scan (used by local, Dropbox, OneDrive, Nextcloud, WebDAV)
 - `delete_after_process` — whether to remove source files after ingestion (default: `false`)
 
-> **Security**: Path traversal protection is enforced on user-configured watch folder paths. Relative paths, `..` components, and symlink escapes are rejected to prevent access to files outside the intended directory.
+Additional type-specific config fields:
+- **S3**: `bucket`, `region`, `prefix`, `endpoint_url`
+- **Google Drive**: `folder_id`
+- **Nextcloud / WebDAV**: `url`, `folder_path`
+
+> **Security**: Path traversal protection is enforced on local watch folder paths. Relative paths, `..` components, and symlink escapes are rejected. Cloud source types use per-user encrypted credentials instead.
 
 - Individual scan failures are handled gracefully and recorded on the integration's `last_error` field without interrupting the scanning of other integrations.
 - The scan runs alongside the system-level watch folder polling cycle.
@@ -302,7 +308,15 @@ DocuElevate can automatically pull document attachments from IMAP mailboxes —
 
 In addition to system-level mailboxes, each user can configure personal IMAP sources through the **Integrations** dashboard (`/integrations`). Documents ingested from per-user IMAP integrations are automatically attributed to the owning user's `owner_id`.
 
-Per-user IMAP integrations are stored in the `user_integrations` table with `integration_type='IMAP'` and `direction='SOURCE'`. Credentials are encrypted at rest using Fernet encryption.
+Per-user IMAP integrations are stored in the `user_integrations` table with `integration_type='IMAP'` and `direction='SOURCE'`. The `config` JSON field stores:
+- `host` — IMAP server hostname (required)
+- `port` — IMAP server port (default: `993`)
+- `username` — IMAP login username (required)
+- `use_ssl` — whether to use SSL/TLS (default: `true`)
+- `delete_after_process` — whether to delete emails from the mailbox after processing (default: `false`)
+- `gmail_apply_labels` — whether to apply Gmail-specific labels and stars to processed emails (default: `true`). When enabled, processed emails are starred and tagged with an "Ingested" label. Only applies to Gmail hosts.
+
+Credentials are encrypted at rest using Fernet encryption.
 
 - Individual connection failures are handled gracefully and recorded on the integration's `last_error` field without interrupting the polling of other integrations.
 - The polling loop runs every minute and processes all active IMAP sources (system-level and per-user) in sequence.

docs/UserGuide.md

@@ -157,13 +157,13 @@ The **Integrations** page (`/integrations`) provides a unified view of all your
 2. Choose a **Direction** — Source (ingestion) or Destination (storage).
 3. Choose an **Integration Type** (e.g. IMAP, S3, Dropbox, WebDAV).
 4. Fill in the type-specific fields — the form adapts dynamically based on your choice:
-   - **IMAP** — host, port, username, password, SSL toggle
+   - **IMAP** — host, port, username, password, SSL toggle, delete after processing, Gmail labels & star toggle
    - **S3** — bucket, region, access key, secret key
    - **WebDAV / Nextcloud** — URL, folder, username, password
    - **FTP / SFTP** — host, port, remote path, username, password
    - **Dropbox / Google Drive / OneDrive** — folder path, with a link to the OAuth setup page
    - **Email Forward** — recipient email address
-   - **Watch Folder** — folder path
+   - **Watch Folder** — source type (Local, S3, Dropbox, Google Drive, OneDrive, Nextcloud, WebDAV), per-type config fields, delete after processing toggle
    - **Paperless NGX** — URL and API token
    - **Webhook** — no configuration needed; the form shows a quick-start guide with sample `curl` and Python snippets for uploading documents via the API
 5. Click **Test Connection** to verify the settings before saving.

frontend/templates/integrations_dashboard.html

@@ -461,16 +461,28 @@
               <label for="imap-password" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Password <span class="text-red-500" aria-hidden="true">*</span></label>
               <input id="imap-password" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.password" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
             </div>
-            <div class="flex items-center gap-4">
+            <div class="flex flex-wrap items-center gap-4">
               <label class="inline-flex items-center gap-2 text-sm text-gray-700 dark:text-gray-200 cursor-pointer">
                 <input type="checkbox" x-model="form.config.use_ssl" class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500" />
                 Use SSL
               </label>
+              <label class="inline-flex items-center gap-2 text-sm text-gray-700 dark:text-gray-200 cursor-pointer">
+                <input type="checkbox" x-model="form.config.delete_after_process" class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500" />
+                Delete emails after processing
+              </label>
+              <label class="inline-flex items-center gap-2 text-sm text-gray-700 dark:text-gray-200 cursor-pointer">
+                <input type="checkbox" x-model="form.config.gmail_apply_labels" class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500" />
+                Apply Gmail labels &amp; star
+              </label>
               <label class="inline-flex items-center gap-2 text-sm text-gray-700 dark:text-gray-200 cursor-pointer">
                 <input type="checkbox" x-model="showPassword" class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500" />
                 Show password
               </label>
             </div>
+            <p class="text-xs text-gray-500 dark:text-gray-400">
+              <i class="fas fa-info-circle mr-1" aria-hidden="true"></i>
+              Gmail labels &amp; star: when enabled, processed emails are starred and tagged with an "Ingested" label in Gmail. Only applies to Gmail servers.
+            </p>
           </div>
         </template>
 
@@ -603,8 +615,182 @@
           <div class="space-y-3 border-t border-gray-200 dark:border-gray-700 pt-3">
             <p class="text-xs font-semibold text-gray-400 uppercase tracking-wider">Watch Folder Settings</p>
             <div>
-              <label for="wf-path" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder Path <span class="text-red-500" aria-hidden="true">*</span></label>
+              <label for="wf-source-type" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Source Type <span class="text-red-500" aria-hidden="true">*</span></label>
-              <input id="wf-path" type="text" x-model="form.config.path" placeholder="/data/inbox" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+              <select id="wf-source-type" x-model="form.config.source_type" @change="onWatchFolderSourceTypeChange()" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true">
+                <option value="local">Local Filesystem</option>
+                <option value="s3">Amazon S3</option>
+                <option value="dropbox">Dropbox</option>
+                <option value="google_drive">Google Drive</option>
+                <option value="onedrive">OneDrive</option>
+                <option value="nextcloud">Nextcloud</option>
+                <option value="webdav">WebDAV</option>
+              </select>
+            </div>
+
+            <!-- Local filesystem fields -->
+            <template x-if="form.config.source_type === 'local'">
+              <div>
+                <label for="wf-path" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder Path <span class="text-red-500" aria-hidden="true">*</span></label>
+                <input id="wf-path" type="text" x-model="form.config.folder_path" placeholder="/data/inbox" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+              </div>
+            </template>
+
+            <!-- S3 fields -->
+            <template x-if="form.config.source_type === 's3'">
+              <div class="space-y-3">
+                <div class="grid grid-cols-2 gap-3">
+                  <div>
+                    <label for="wf-s3-bucket" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Bucket <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-s3-bucket" type="text" x-model="form.config.bucket" placeholder="my-bucket" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                  </div>
+                  <div>
+                    <label for="wf-s3-region" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Region</label>
+                    <input id="wf-s3-region" type="text" x-model="form.config.region" placeholder="us-east-1" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                  </div>
+                </div>
+                <div>
+                  <label for="wf-s3-prefix" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Prefix (folder path)</label>
+                  <input id="wf-s3-prefix" type="text" x-model="form.config.prefix" placeholder="inbox/" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                </div>
+                <div>
+                  <label for="wf-s3-endpoint" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Endpoint URL <span class="text-xs text-gray-400">(optional, for S3-compatible)</span></label>
+                  <input id="wf-s3-endpoint" type="text" x-model="form.config.endpoint_url" placeholder="https://s3.example.com" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                </div>
+                <div class="grid grid-cols-2 gap-3">
+                  <div>
+                    <label for="wf-s3-ak" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Access Key ID <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-s3-ak" type="text" x-model="form.credentials.access_key_id" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                  </div>
+                  <div>
+                    <label for="wf-s3-sk" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Secret Access Key <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-s3-sk" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.secret_access_key" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                  </div>
+                </div>
+              </div>
+            </template>
+
+            <!-- Dropbox fields -->
+            <template x-if="form.config.source_type === 'dropbox'">
+              <div class="space-y-3">
+                <div>
+                  <label for="wf-dbx-folder" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder Path <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-dbx-folder" type="text" x-model="form.config.folder_path" placeholder="/Inbox/Scanner" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div>
+                  <label for="wf-dbx-token" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Refresh Token <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-dbx-token" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.refresh_token" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                </div>
+                <div class="grid grid-cols-2 gap-3">
+                  <div>
+                    <label for="wf-dbx-key" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">App Key <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-dbx-key" type="text" x-model="form.credentials.app_key" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                  </div>
+                  <div>
+                    <label for="wf-dbx-secret" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">App Secret <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-dbx-secret" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.app_secret" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                  </div>
+                </div>
+              </div>
+            </template>
+
+            <!-- Google Drive fields -->
+            <template x-if="form.config.source_type === 'google_drive'">
+              <div class="space-y-3">
+                <div>
+                  <label for="wf-gd-folder" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder ID <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-gd-folder" type="text" x-model="form.config.folder_id" placeholder="1ABCdef..." required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div>
+                  <label for="wf-gd-creds" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Service Account JSON <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <textarea id="wf-gd-creds" x-model="form.credentials.credentials_json" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" rows="3" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm font-mono"></textarea>
+                </div>
+              </div>
+            </template>
+
+            <!-- OneDrive fields -->
+            <template x-if="form.config.source_type === 'onedrive'">
+              <div class="space-y-3">
+                <div>
+                  <label for="wf-od-folder" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder Path <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-od-folder" type="text" x-model="form.config.folder_path" placeholder="/Documents/Inbox" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div>
+                  <label for="wf-od-token" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Refresh Token <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-od-token" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.refresh_token" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                </div>
+                <div class="grid grid-cols-2 gap-3">
+                  <div>
+                    <label for="wf-od-cid" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Client ID <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-od-cid" type="text" x-model="form.credentials.client_id" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                  </div>
+                  <div>
+                    <label for="wf-od-cs" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Client Secret <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-od-cs" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.client_secret" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                  </div>
+                </div>
+              </div>
+            </template>
+
+            <!-- Nextcloud fields -->
+            <template x-if="form.config.source_type === 'nextcloud'">
+              <div class="space-y-3">
+                <div>
+                  <label for="wf-nc-url" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Nextcloud URL <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-nc-url" type="url" x-model="form.config.url" placeholder="https://cloud.example.com" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div>
+                  <label for="wf-nc-folder" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder Path <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-nc-folder" type="text" x-model="form.config.folder_path" placeholder="/Documents/Inbox" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div class="grid grid-cols-2 gap-3">
+                  <div>
+                    <label for="wf-nc-user" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Username <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-nc-user" type="text" x-model="form.credentials.username" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                  </div>
+                  <div>
+                    <label for="wf-nc-pass" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Password <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-nc-pass" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.password" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                  </div>
+                </div>
+              </div>
+            </template>
+
+            <!-- WebDAV fields -->
+            <template x-if="form.config.source_type === 'webdav'">
+              <div class="space-y-3">
+                <div>
+                  <label for="wf-dav-url" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">WebDAV URL <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-dav-url" type="url" x-model="form.config.url" placeholder="https://webdav.example.com/dav/" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div>
+                  <label for="wf-dav-folder" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Folder Path <span class="text-red-500" aria-hidden="true">*</span></label>
+                  <input id="wf-dav-folder" type="text" x-model="form.config.folder_path" placeholder="/remote.php/webdav/Inbox" required class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                </div>
+                <div class="grid grid-cols-2 gap-3">
+                  <div>
+                    <label for="wf-dav-user" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Username <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-dav-user" type="text" x-model="form.credentials.username" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" aria-required="true" />
+                  </div>
+                  <div>
+                    <label for="wf-dav-pass" class="block text-sm font-medium text-gray-700 dark:text-gray-200 mb-1">Password <span class="text-red-500" aria-hidden="true">*</span></label>
+                    <input id="wf-dav-pass" :type="showPassword ? 'text' : 'password'" x-model="form.credentials.password" :placeholder="editingIntegration ? '(unchanged if blank)' : ''" :required="!editingIntegration" class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-2 focus:ring-indigo-500 dark:bg-gray-700 dark:text-white text-sm" />
+                  </div>
+                </div>
+              </div>
+            </template>
+
+            <!-- Common watch folder options -->
+            <div class="flex flex-wrap items-center gap-4 pt-2">
+              <label class="inline-flex items-center gap-2 text-sm text-gray-700 dark:text-gray-200 cursor-pointer">
+                <input type="checkbox" x-model="form.config.delete_after_process" class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500" />
+                Delete files after processing
+              </label>
+              <template x-if="form.config.source_type !== 'local'">
+                <label class="inline-flex items-center gap-2 text-sm text-gray-700 dark:text-gray-200 cursor-pointer">
+                  <input type="checkbox" x-model="showPassword" class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500" />
+                  Show secrets
+                </label>
+              </template>
             </div>
           </div>
         </template>
@@ -995,7 +1181,7 @@ function integrationsDashboard() {
       this.testResult = null;
       // Set sensible defaults
       if (this.form.integration_type === 'IMAP') {
-        this.form.config = { host: '', port: 993, username: '', use_ssl: true };
+        this.form.config = { host: '', port: 993, username: '', use_ssl: true, delete_after_process: false, gmail_apply_labels: true };
         this.form.credentials = { password: '' };
       } else if (this.form.integration_type === 'S3') {
         this.form.config = { bucket: '', region: 'us-east-1', folder_prefix: '', endpoint_url: '' };
@@ -1013,7 +1199,7 @@ function integrationsDashboard() {
         this.form.config = { recipient: '' };
         this.form.credentials = {};
       } else if (this.form.integration_type === 'WATCH_FOLDER') {
-        this.form.config = { path: '' };
+        this.form.config = { source_type: 'local', folder_path: '', delete_after_process: false };
         this.form.credentials = {};
       } else if (this.form.integration_type === 'PAPERLESS') {
         this.form.config = { url: '' };
@@ -1021,6 +1207,33 @@ function integrationsDashboard() {
       }
     },
 
+    onWatchFolderSourceTypeChange() {
+      const st = this.form.config.source_type;
+      const dap = this.form.config.delete_after_process || false;
+      this.form.credentials = {};
+      if (st === 'local') {
+        this.form.config = { source_type: st, folder_path: '', delete_after_process: dap };
+      } else if (st === 's3') {
+        this.form.config = { source_type: st, bucket: '', region: 'us-east-1', prefix: '', endpoint_url: '', delete_after_process: dap };
+        this.form.credentials = { access_key_id: '', secret_access_key: '' };
+      } else if (st === 'dropbox') {
+        this.form.config = { source_type: st, folder_path: '', delete_after_process: dap };
+        this.form.credentials = { refresh_token: '', app_key: '', app_secret: '' };
+      } else if (st === 'google_drive') {
+        this.form.config = { source_type: st, folder_id: '', delete_after_process: dap };
+        this.form.credentials = { credentials_json: '' };
+      } else if (st === 'onedrive') {
+        this.form.config = { source_type: st, folder_path: '', delete_after_process: dap };
+        this.form.credentials = { refresh_token: '', client_id: '', client_secret: '' };
+      } else if (st === 'nextcloud') {
+        this.form.config = { source_type: st, url: '', folder_path: '', delete_after_process: dap };
+        this.form.credentials = { username: '', password: '' };
+      } else if (st === 'webdav') {
+        this.form.config = { source_type: st, url: '', folder_path: '', delete_after_process: dap };
+        this.form.credentials = { username: '', password: '' };
+      }
+    },
+
     openCreateModal() {
       this.editingIntegration = null;
       this.form = {

tests/test_imap_tasks.py

@@ -687,6 +687,59 @@ class TestPullInbox:
         mock_star.assert_called_once_with(mock_mail, b"1")
         mock_label.assert_called_once_with(mock_mail, b"1", label="Ingested")
 
+    @patch("app.tasks.imap_tasks.email_already_has_label")
+    @patch("app.tasks.imap_tasks.mark_as_processed_with_label")
+    @patch("app.tasks.imap_tasks.mark_as_processed_with_star")
+    @patch("app.tasks.imap_tasks.fetch_attachments_and_enqueue")
+    @patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
+    @patch("app.tasks.imap_tasks.load_processed_emails")
+    @patch("app.tasks.imap_tasks.save_processed_emails")
+    @patch("app.tasks.imap_tasks.settings")
+    def test_gmail_labels_disabled_when_gmail_apply_labels_false(
+        self,
+        mock_settings,
+        mock_save,
+        mock_load,
+        mock_imap_class,
+        mock_fetch,
+        mock_star,
+        mock_label,
+        mock_has_label,
+    ):
+        """Gmail star/label operations should be skipped when gmail_apply_labels=False."""
+        mock_settings.workdir = "/tmp"
+        mock_settings.imap_readonly_mode = False
+        mock_load.return_value = {}
+        mock_mail = MagicMock()
+        mock_imap_class.return_value = mock_mail
+
+        import email
+
+        msg = email.message.EmailMessage()
+        msg["Message-ID"] = "<test-no-labels@gmail.com>"
+        raw_email = msg.as_bytes()
+
+        mock_mail.login.return_value = ("OK", [])
+        mock_mail.select.return_value = ("OK", [])
+        mock_mail.search.return_value = ("OK", [b"1"])
+        mock_mail.fetch.return_value = ("OK", [[None, raw_email]])
+
+        pull_inbox(
+            mailbox_key="imap2",
+            host="imap.gmail.com",
+            port=993,
+            username="user@gmail.com",
+            password=_TEST_CREDENTIAL,
+            use_ssl=True,
+            delete_after_process=False,
+            gmail_apply_labels=False,
+        )
+
+        mock_star.assert_not_called()
+        mock_label.assert_not_called()
+        mock_has_label.assert_not_called()
+        mock_fetch.assert_called()
+
     @patch("app.tasks.imap_tasks.email_already_has_label")
     @patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
     @patch("app.tasks.imap_tasks.load_processed_emails")
@@ -1531,6 +1584,55 @@ class TestPullUserIntegrationImap:
         assert mock_integ.last_error is None
         assert mock_integ.last_used_at is not None
 
+    @patch("app.tasks.imap_tasks._get_db_session")
+    @patch("app.tasks.imap_tasks.pull_inbox")
+    def test_passes_gmail_apply_labels_config_to_pull_inbox(self, mock_pull, mock_session_factory):
+        """gmail_apply_labels config should be forwarded to pull_inbox."""
+        from app.tasks.imap_tasks import _pull_user_integration_imap
+
+        mock_integ = MagicMock()
+        mock_integ.id = 14
+        mock_integ.owner_id = "owner-gmail"
+        mock_integ.config = (
+            '{"host": "imap.gmail.com", "port": 993, "username": "u@gmail.com",'
+            ' "use_ssl": true, "gmail_apply_labels": false}'
+        )
+        mock_integ.credentials = "enc:encrypted"
+
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.all.return_value = [mock_integ]
+        mock_session_factory.return_value = mock_db
+
+        with patch("app.utils.encryption.decrypt_value", return_value='{"password": "p"}'):
+            _pull_user_integration_imap()
+
+        mock_pull.assert_called_once()
+        call_kwargs = mock_pull.call_args
+        assert call_kwargs.kwargs.get("gmail_apply_labels") is False
+
+    @patch("app.tasks.imap_tasks._get_db_session")
+    @patch("app.tasks.imap_tasks.pull_inbox")
+    def test_gmail_apply_labels_defaults_to_true(self, mock_pull, mock_session_factory):
+        """Config without gmail_apply_labels should default to True."""
+        from app.tasks.imap_tasks import _pull_user_integration_imap
+
+        mock_integ = MagicMock()
+        mock_integ.id = 15
+        mock_integ.owner_id = "owner-gmail2"
+        mock_integ.config = '{"host": "imap.gmail.com", "port": 993, "username": "u@gmail.com", "use_ssl": true}'
+        mock_integ.credentials = "enc:encrypted"
+
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.all.return_value = [mock_integ]
+        mock_session_factory.return_value = mock_db
+
+        with patch("app.utils.encryption.decrypt_value", return_value='{"password": "p"}'):
+            _pull_user_integration_imap()
+
+        mock_pull.assert_called_once()
+        call_kwargs = mock_pull.call_args
+        assert call_kwargs.kwargs.get("gmail_apply_labels") is True
+
 
 @pytest.mark.unit
 class TestPullAllInboxesCallsIntegrations: