fix: switch download request to be async with firestore push updates

Closes #599

.devcontainer/devcontainer.json

@@ -4,7 +4,6 @@
     "target": "dev"
   },
   "containerEnv": {
-    "STORAGE_EMULATOR_HOST": "http://host.docker.internal:9199",
     "GOOGLE_APPLICATION_CREDENTIALS": "/gcp/config/application_default_credentials.json",
     "CLOUDSDK_CONFIG": "/gcp/config/application_default_credentials.json"
   },
@@ -28,8 +27,5 @@
       "type": "bind"
     }
   ],
-  "postCreateCommand": "cd cloudrun && ./dev.sh",
-  "runArgs": [
-    "--network=host" // allows the container to access the host's network
-  ]
+  "postCreateCommand": "cd cloudrun && ./dev.sh"
 }

.storybook/preview.jsx

@@ -1,7 +1,7 @@
 import '@arcgis/core/assets/esri/themes/light/main.css';
 import '@utahdts/utah-design-system-header/css';
 import '../src/index.css';
-import { RemoteConfigContext } from '../src/RemoteConfigProvider';
+import { RemoteConfigContext } from '../src/contexts/RemoteConfigProvider';
 import remoteConfigDefaultJson from '../src/remote_config_defaults.json';
 
 /** @type {import('@storybook/react').Preview} */

.vscode/settings.json

@@ -8,6 +8,7 @@
     "azuretools",
     "bfnontargeted",
     "bftargeted",
+    "Brownfields",
     "builtins",
     "ca",
     "CERCLABRANCHACTMAJ",
@@ -16,6 +17,7 @@
     "charliermarsh",
     "checkmark",
     "cloudrun",
+    "CLOUDSDK",
     "DAGGETT",
     "dataframe",
     "datepicker",
@@ -27,6 +29,7 @@
     "DERRID",
     "dischargers",
     "dotenv",
+    "dowork",
     "DUCHESNE",
     "dwqnpdes",
     "dwqnpdes dischargers",
@@ -72,6 +75,7 @@
     "nodebuffer",
     "noninteractive",
     "opensgid",
+    "osgeo",
     "overscan",
     "packagejson",
     "permissionproxy",

cloudrun/setup.py

@@ -41,6 +41,7 @@
     install_requires=[
         "google-cloud-logging==3.*",
         "google-cloud-storage==2.*",
+        "google-cloud-firestore==2.*",
         "flask-cors==4.*",
         "flask-json==0.4",
         "flask==3.*",

cloudrun/src/download/agol.py

@@ -11,6 +11,7 @@
 from osgeo import gdal, ogr
 
 from .log import logger
+from .database import update_job_layer
 
 #: throw exceptions on errors rather than returning None
 gdal.UseExceptions()
@@ -60,7 +61,7 @@ def write_to_output(tableName, feature_set, format):
         raise ValueError(f"unsupported format: {format}")
 
 
-def download(layers, format):
+def download(id, layers, format):
     cleanup()
 
     return_geometry = format in formats_with_shape
@@ -72,16 +73,31 @@ def download(layers, format):
         tableName = layer["tableName"]
         logger.info(f"query layer: {tableName}")
 
-        feature_set = get_agol_data(layer["url"], layer["objectIds"], return_geometry)
+        success = False
+        try:
+            feature_set = get_agol_data(
+                layer["url"], layer["objectIds"], return_geometry
+            )
 
-        if len(feature_set.features) == 0:
-            logger.info("no features found, skipping creation")
+            if len(feature_set.features) == 0:
+                logger.info("no features found, skipping creation")
 
-            continue
+                update_job_layer(id, tableName, True)
 
-        write_to_output(tableName, feature_set, format)
+                continue
 
-        if "relationships" in layer and len(layer["relationships"]) > 0:
+            write_to_output(tableName, feature_set, format)
+
+            update_job_layer(id, tableName, True)
+
+            success = True
+        except Exception as e:
+            logger.info(f"error processing layer: {tableName}")
+            logger.error(e)
+
+            update_job_layer(id, tableName, False, str(e))
+
+        if success and "relationships" in layer and len(layer["relationships"]) > 0:
             primary_key = layer["relationships"][0]["primary"]
             primary_keys = feature_set.sdf.reset_index()[primary_key].tolist()
 

cloudrun/src/download/bucket.py

@@ -3,14 +3,13 @@
 """
 from google.cloud import storage
 import shutil
-from uuid import uuid4
 from .log import logger
 
 storage_client = storage.Client()
 bucket = storage_client.bucket("ut-dts-agrc-deq-enviro-dev.appspot.com")
 
 
-def upload(path):
+def upload(id, path):
     """
     Uploads the data to a cloud storage bucket.
     """
@@ -19,7 +18,6 @@ def upload(path):
     # zip contents of path
     zip_file = shutil.make_archive(path, "zip", path)
 
-    id = uuid4().hex
     blob = bucket.blob(f"{id}.zip")
     blob.upload_from_filename(zip_file)
 

cloudrun/src/download/database.py

@@ -0,0 +1,81 @@
+"""
+A module for interacting with firestore
+"""
+from os import environ
+from uuid import uuid4
+
+from google.cloud import firestore
+
+project = None
+if environ.get("FLASK_DEBUG") == "1":
+    #: setting the GOOGLE_CLOUD_PROJECT env var, causes gcloud to attempt to reauth
+    project = "ut-dts-agrc-deq-enviro-dev"
+
+client = firestore.Client(project)
+
+"""
+Job document structure:
+{
+  "id": "string",
+  "created": "timestamp",
+  "updated": "timestamp",
+  "status": "processing|complete|failed",
+  "format": "csv|excel|filegdb|geojson|shapefile",
+  "layers": {
+    "tableName": {
+      "processed": "boolean",
+      "error": "string"
+    }
+  },
+  "error": "string",
+}
+"""
+
+
+def create_job(layers, format):
+    """
+    Creates a job document in firestore.
+    """
+    id = uuid4().hex
+
+    doc_ref = client.collection("jobs").document(id)
+    doc_ref.set(
+        {
+            "id": id,
+            "created": firestore.SERVER_TIMESTAMP,
+            "updated": firestore.SERVER_TIMESTAMP,
+            "status": "processing",
+            "format": format,
+            "layers": {layer: {"error": None, "processed": False} for layer in layers},
+        }
+    )
+
+    return id
+
+
+def update_job_status(id, status, error=None):
+    """
+    Updates a job document in firestore.
+    """
+    doc_ref = client.collection("jobs").document(id)
+    doc_ref.update(
+        {
+            "updated": firestore.SERVER_TIMESTAMP,
+            "status": status,
+            "error": error,
+        }
+    )
+
+
+def update_job_layer(id, layer, processed, error=None):
+    """
+    Updates a job document in firestore.
+    """
+    doc_ref = client.collection("jobs").document(id)
+    doc_ref.update(
+        {
+            "updated": firestore.SERVER_TIMESTAMP,
+            f"layers.{layer}.processed": processed,
+            f"layers.{layer}.error": error,
+        }
+    )

cloudrun/src/download/main.py

@@ -1,14 +1,17 @@
 """
 Download server.
 """
-from flask import Flask, request
-from .agol import download, cleanup
-from . import bucket
-from dotenv import load_dotenv
+import threading
+import traceback
 
+from dotenv import load_dotenv
+from flask import Flask, request
 from flask_cors import CORS
 from flask_json import FlaskJSON
 
+from . import bucket, database, log
+from .agol import cleanup, download
+
 load_dotenv()
 
 formats = [
@@ -25,6 +28,23 @@
 CORS(app)
 
 
+def dowork(id, layers, format):
+    log.logger.info(f"Starting job {id}")
+    try:
+        output_path = download(id, layers, format)
+        bucket.upload(id, output_path)
+
+        database.update_job_status(id, "complete")
+    except Exception as e:
+        # Print stack trace to log
+        log.logger.error(traceback.format_exc())
+        database.update_job_status(id, "failed", str(e))
+    finally:
+        cleanup()
+
+    log.logger.info(f"Job {id} complete")
+
+
 @app.post("/generate")
 def generate():
     """
@@ -37,13 +57,16 @@ def generate():
         return {"success": False, "error": f"invalid format value: {format}"}, 400
 
     try:
-        output_path = download(layers, format)
+        id = database.create_job([layer["tableName"] for layer in layers], format)
 
-        id = bucket.upload(output_path)
-    finally:
-        cleanup()
+        # do the work in a separate thread so we can return the id right away
+        thread = threading.Thread(target=dowork, args=(id, layers, format))
+        thread.start()
 
-    return {"id": id, "success": True}
+        return {"id": id, "success": True}
+    except Exception as e:
+        log.logger.error(e)
+        return {"success": False, "error": str(e)}, 500
 
 
 @app.get("/download/<id>/data.zip")

firebase.json

@@ -123,6 +123,10 @@
     }
   },
   "storage": {
-    "rules": "storage.rules"
+    "rules": "firebase.rules"
+  },
+  "firestore": {
+    "rules": "firebase.rules",
+    "indexes": "firestore.indexes.json"
   }
 }

firebase.rules

@@ -0,0 +1,17 @@
+rules_version = '2';
+
+service cloud.firestore {
+  match /databases/{database}/documents {
+    match /{document=**} {
+      allow read: if true;
+    }
+  }
+}
+
+service firebase.storage {
+  match /b/{bucket}/o {
+    match /{allPaths=**} {
+      allow read: if true;
+    }
+  }
+}

firestore.indexes.json

@@ -0,0 +1,4 @@
+{
+  "indexes": [],
+  "fieldOverrides": []
+}