Add basic MEAD artist profile updates

fixtures/mead_artist_update.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<mead:MeadMessage
+  xmlns:mead="http://ddex.net/xml/mead/10"
+  AvsVersionId="5">
+  <MessageHeader>
+    <MessageId>mead-artist-update-1</MessageId>
+    <MessageCreatedDateTime>2026-06-22T12:00:00Z</MessageCreatedDateTime>
+  </MessageHeader>
+  <PartyInformationList>
+    <PartyInformation>
+      <PartySummary>
+        <PartyReference>P1</PartyReference>
+        <PartyName>
+          <FullName>
+            <Name>DJ Theo</Name>
+          </FullName>
+        </PartyName>
+      </PartySummary>
+      <Pseudonym>
+        <Name>
+          <FullName>
+            <Name>DJ Theo Official</Name>
+          </FullName>
+        </Name>
+        <IsOfficial>true</IsOfficial>
+      </Pseudonym>
+      <Biography>
+        <Text>Oakland producer and DJ building vivid left-field dance records.</Text>
+      </Biography>
+      <Image>
+        <File>
+          <URI>resources/Image_001_001.jpg</URI>
+        </File>
+        <ImageType UserDefinedValue="ProfilePicture" />
+      </Image>
+    </PartyInformation>
+  </PartyInformationList>
+</mead:MeadMessage>

src/db.ts

@@ -1,5 +1,6 @@
-import { DDEXRelease } from './parseDelivery'
+import type { DDEXArtistProfileUpdate, DDEXRelease } from './parseDelivery'
 
+export { artistProfileUpdateRepo } from './db/artistProfileUpdateRepo'
 export { assetRepo } from './db/assetRepo'
 export { isClearedRepo } from './db/isClearedRepo'
 export { releaseRepo } from './db/releaseRepo'
@@ -34,6 +35,13 @@ export enum ReleaseProcessingStatus {
   Deleted = 'Deleted',
 }
 
+export enum ArtistProfileUpdateStatus {
+  Blocked = 'Blocked',
+  PublishPending = 'PublishPending',
+  Published = 'Published',
+  Failed = 'Failed',
+}
+
 export type ReleaseRow = DDEXRelease & {
   source: string
   key: string
@@ -60,6 +68,21 @@ export type ReleaseRow = DDEXRelease & {
   mediaDeletedAt?: string
 }
 
+export type ArtistProfileUpdateRow = DDEXArtistProfileUpdate & {
+  source: string
+  key: string
+  xmlUrl: string
+  messageTimestamp: string
+  status: ArtistProfileUpdateStatus
+  createdAt: string
+  updatedAt: string
+  publishedAt?: string
+  blockHash?: string
+  blockNumber?: number
+  lastPublishError: string
+  publishErrorCount: number
+}
+
 export type S3MarkerRow = {
   bucket: string
   marker: string

src/db/artistProfileUpdateRepo.ts

@@ -0,0 +1,127 @@
+import { ArtistProfileUpdateRow, ArtistProfileUpdateStatus } from '../db'
+import { DDEXArtistProfileUpdate } from '../parseDelivery'
+import { omitEmpty } from '../util'
+import { ifdef, pgUpdate, pgUpsert, sql } from './sql'
+
+type FindArtistProfileUpdateParams = {
+  pendingPublish?: boolean
+  source?: string
+  limit?: number
+}
+
+export const artistProfileUpdateRepo = {
+  chooseKey(source: string, xmlUrl: string, update: DDEXArtistProfileUpdate) {
+    const id =
+      update.audiusUser ||
+      update.artistHandle ||
+      update.artistName ||
+      update.partyRef
+    if (!id) {
+      const msg = `failed to chooseArtistProfileUpdateKey: ${JSON.stringify(
+        update
+      )}`
+      console.log(msg)
+      throw new Error(msg)
+    }
+    return [source, xmlUrl, id].join(':')
+  },
+
+  async all(params?: FindArtistProfileUpdateParams) {
+    params ||= {}
+    const rows: ArtistProfileUpdateRow[] = await sql`
+      select * from artist_profile_updates
+      where 1=1
+
+      ${ifdef(
+        params.pendingPublish,
+        sql`
+          and status in (
+            ${ArtistProfileUpdateStatus.PublishPending},
+            ${ArtistProfileUpdateStatus.Failed}
+          )
+          and "publishErrorCount" < 5
+        `
+      )}
+
+      ${ifdef(params.source, sql` and "source" = ${params.source!} `)}
+
+      order by "messageTimestamp" asc
+
+      ${ifdef(params.limit, sql` limit ${params.limit!} `)}
+    `
+
+    return rows
+  },
+
+  async get(key: string) {
+    const rows =
+      await sql`select * from artist_profile_updates where "key" = ${key}`
+    const row = rows[0]
+    if (!row) return
+    return row as ArtistProfileUpdateRow
+  },
+
+  async update(r: Partial<ArtistProfileUpdateRow>) {
+    await pgUpdate('artist_profile_updates', 'key', r)
+  },
+
+  async upsert(
+    source: string,
+    xmlUrl: string,
+    messageTimestamp: string,
+    update: DDEXArtistProfileUpdate
+  ) {
+    const key = artistProfileUpdateRepo.chooseKey(source, xmlUrl, update)
+    const prior = await artistProfileUpdateRepo.get(key)
+
+    if (
+      prior &&
+      (prior.messageTimestamp > messageTimestamp ||
+        (prior.messageTimestamp == messageTimestamp &&
+          prior.status == ArtistProfileUpdateStatus.Published))
+    ) {
+      console.log(`skipping ${xmlUrl} because ${key} is newer`)
+      return
+    }
+
+    const status = update.problems.length
+      ? ArtistProfileUpdateStatus.Blocked
+      : ArtistProfileUpdateStatus.PublishPending
+
+    const data = {
+      source,
+      key,
+      status,
+      xmlUrl,
+      messageTimestamp,
+      updatedAt: new Date().toISOString(),
+      ...update,
+    } as Partial<ArtistProfileUpdateRow>
+
+    await pgUpsert('artist_profile_updates', 'key', omitEmpty(data))
+  },
+
+  async addPublishError(key: string, err: Error) {
+    const status = ArtistProfileUpdateStatus.Failed
+    const errText = err.stack || err.toString()
+    await sql`
+      update artist_profile_updates set
+        status=${status},
+        "lastPublishError"=${errText},
+        "publishErrorCount" = "publishErrorCount" + 1
+      where "key" = ${key}
+    `
+  },
+
+  async addPublishBlock(key: string, err: Error) {
+    const status = ArtistProfileUpdateStatus.Blocked
+    const errText = err.stack || err.toString()
+    await sql`
+      update artist_profile_updates set
+        status=${status},
+        "lastPublishError"=${errText},
+        "publishErrorCount" = "publishErrorCount" + 1
+      where "key" = ${key}
+    `
+  },
+}

src/db/migrations.ts

@@ -156,6 +156,34 @@ const steps = [
   sql`ALTER TABLE releases ADD COLUMN IF NOT EXISTS "plannedEntityId" text;`,
   sql`ALTER TABLE releases ADD COLUMN IF NOT EXISTS "plannedTrackIds" jsonb;`,
   sql`ALTER TABLE releases ADD COLUMN IF NOT EXISTS "partialTrackIds" jsonb;`,
+
+  sql`
+  create table if not exists artist_profile_updates (
+    "source" text not null,
+    "key" text primary key,
+    "xmlUrl" text not null,
+    "messageTimestamp" text,
+    "partyRef" text,
+    "artistName" text,
+    "artistHandle" text,
+    "audiusUser" text,
+    "displayName" text,
+    "bio" text,
+    "profilePicture" jsonb,
+    "coverArt" jsonb,
+    "problems" jsonb,
+    "status" text not null,
+    "blockHash" text,
+    "blockNumber" integer,
+    "publishedAt" timestamptz,
+    "publishErrorCount" integer default 0,
+    "lastPublishError" text,
+    "createdAt" timestamptz DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" timestamptz
+  );
+  `,
+
+  sql`CREATE INDEX IF NOT EXISTS idx_artist_profile_updates_pending ON artist_profile_updates ("status", "publishErrorCount");`,
 ]
 
 // poor man's migrate

src/db/userRepo.ts

@@ -23,6 +23,14 @@ export const userRepo = {
     return users[0]
   },
 
+  async findByHandleAndApiKey(handle: string, apiKey: string) {
+    const users: UserRow[] = await sql`
+      select * from users
+      where lower(handle) = lower(${handle}) and "apiKey" = ${apiKey}
+    `
+    return users[0]
+  },
+
   async upsert(user: UserRow) {
     await sql`
       insert into users ${sql(user)}