Update links in records published to post-ingest SNS topic

Gives subscribers to the post-ingest topic the full URIs of the ingested
collections and items by repurposing the link generation fucntions used by the
API.

Author: jwalgran

serverless.example.yml

@@ -22,6 +22,9 @@ provider:
     # PRE_HOOK: ${self:service}-${self:provider.stage}-preHook
     # API_KEYS_SECRET_ID: ${self:service}-${self:provider.stage}-api-keys
     # POST_HOOK: ${self:service}-${self:provider.stage}-postHook
+    # If you will be subscribing to post-ingest SNS notifications make
+    # sure that STAC_API_URL is set so that links are updated correctly
+    STAC_API_URL: "https://some-stac-server.com"
   iam:
     role:
       statements:

src/lib/api.js

@@ -383,7 +383,7 @@ export const parsePath = function (inpath) {
 }
 
 // Impure - mutates results
-const addCollectionLinks = function (results, endpoint) {
+export const addCollectionLinks = function (results, endpoint) {
   results.forEach((result) => {
     const { id } = result
     let { links } = result
@@ -438,7 +438,7 @@ const addCollectionLinks = function (results, endpoint) {
 }
 
 // Impure - mutates results
-const addItemLinks = function (results, endpoint) {
+export const addItemLinks = function (results, endpoint) {
   results.forEach((result) => {
     let { links } = result
     const { id, collection } = result

src/lib/ingest.js

@@ -1,4 +1,5 @@
 import { getItemCreated } from './database.js'
+import { addItemLinks, addCollectionLinks } from './api.js'
 import { dbClient, createIndex } from './databaseClient.js'
 import logger from './logger.js'
 import { publishRecordToSns } from './sns.js'
@@ -13,6 +14,8 @@ export class InvalidIngestError extends Error {
   }
 }
 
+const hierarchyLinks = ['self', 'root', 'parent', 'child', 'collection', 'item', 'items']
+
 export async function convertIngestObjectToDbObject(
   // eslint-disable-next-line max-len
   /** @type {{ hasOwnProperty: (arg0: string) => any; type: string, collection: string; links: any[]; id: any; }} */ data
@@ -30,12 +33,11 @@ export async function convertIngestObjectToDbObject(
   }
 
   // remove any hierarchy links in a non-mutating way
-  const hlinks = ['self', 'root', 'parent', 'child', 'collection', 'item', 'items']
   if (!data.links) {
     throw new InvalidIngestError('Expected a "links" proporty on the stac object')
   }
   const links = data.links.filter(
-    (/** @type {{ rel: string; }} */ link) => !hlinks.includes(link.rel)
+    (/** @type {{ rel: string; }} */ link) => !hierarchyLinks.includes(link.rel)
   )
   const dbDataObject = { ...data, links }
 
@@ -162,8 +164,34 @@ export async function ingestItems(items) {
   return results
 }
 
+// Impure - mutates record
+function updateLinksWithinRecord(record) {
+  const endpoint = process.env['STAC_API_URL']
+  if (!endpoint) {
+    logger.info('STAC_API_URL not set, not updating links within ingested record')
+    return record
+  }
+  if (!isItem(record) && !isCollection(record)) {
+    logger.info('Record is not a collection or item, not updating links within ingested record')
+    return record
+  }
+
+  record.links = record.links.filter(
+    (/** @type {{ rel: string; }} */ link) => !hierarchyLinks.includes(link.rel)
+  )
+  if (isItem(record)) {
+    addItemLinks([record], endpoint)
+  } else if (isCollection(record)) {
+    addCollectionLinks([record], endpoint)
+  }
+  return record
+}
+
 export async function publishResultsToSns(results, topicArn) {
   results.forEach((result) => {
+    if (result.record && !result.error) {
+      updateLinksWithinRecord(result.record)
+    }
     publishRecordToSns(topicArn, result.record, result.error)
   })
 }

tests/system/test-ingest.js

@@ -1,5 +1,6 @@
 // @ts-nocheck
 
+import url from 'url'
 import test from 'ava'
 import nock from 'nock'
 import { DateTime } from 'luxon'
@@ -349,6 +350,28 @@ test('Ingested collection is published to post-ingest SNS topic', async (t) => {
   t.is(attrs.recordType.Value, 'Collection')
 })
 
+test('Ingested collection is published to post-ingest SNS topic with updated links', async (t) => {
+  const envBeforeTest = { ...process.env }
+  try {
+    const hostname = 'some-stac-server.com'
+    const endpoint = `https://${hostname}`
+    process.env['STAC_API_URL'] = endpoint
+
+    const collection = await loadFixture(
+      'landsat-8-l1-collection.json',
+      { id: randomId('collection') }
+    )
+
+    const { message } = await testPostIngestSNS(t, collection)
+
+    t.truthy(message.record.links)
+    t.true(message.record.links.every((/** @type {Link} */ link) => (
+      link.href && url.parse(link.href).hostname === hostname)))
+  } finally {
+    process.env = envBeforeTest
+  }
+})
+
 test('Ingest collection failure is published to post-ingest SNS topic', async (t) => {
   const { message, attrs } = await testPostIngestSNS(t, {
     type: 'Collection',
@@ -407,3 +430,49 @@ test('Ingest item failure is published to post-ingest SNS topic', async (t) => {
   t.is(attrs.ingestStatus.Value, 'failed')
   t.is(attrs.recordType.Value, 'Item')
 })
+
+test('Ingested item is published to post-ingest SNS topic with updated links', async (t) => {
+  const envBeforeTest = { ...process.env }
+  try {
+    const hostname = 'some-stac-server.com'
+    const endpoint = `https://${hostname}`
+    process.env['STAC_API_URL'] = endpoint
+
+    const collection = await ingestCollectionAndPurgePostIngestQueue(t)
+
+    const item = await loadFixture(
+      'stac/ingest-item.json',
+      { id: randomId('item'), collection: collection.id }
+    )
+
+    const { message } = await testPostIngestSNS(t, item)
+
+    t.truthy(message.record.links)
+    t.true(message.record.links.every((/** @type {Link} */ link) => (
+      link.href && url.parse(link.href).hostname === hostname)))
+  } finally {
+    process.env = envBeforeTest
+  }
+})
+
+test('Ingested item facilure is published to post-ingest SNS topic without updated links', async (t) => {
+  const envBeforeTest = { ...process.env }
+  try {
+    const hostname = 'some-stac-server.com'
+    const endpoint = `https://${hostname}`
+    process.env['STAC_API_URL'] = endpoint
+
+    const item = await loadFixture(
+      'stac/ingest-item.json',
+      { id: randomId('item'), collection: 'INVALID COLLECTION' }
+    )
+
+    const { message } = await testPostIngestSNS(t, item)
+
+    t.truthy(message.record.links)
+    t.false(message.record.links.every((/** @type {Link} */ link) => (
+      link.href && url.parse(link.href).hostname === hostname)))
+  } finally {
+    process.env = envBeforeTest
+  }
+})