Add interactivity to search

packages/client/src/Dashboard.tsx

@@ -7,16 +7,17 @@ import {
   Drawer,
   IconButton,
   InputBase,
+  LinearProgress,
   List,
   ListItem,
   ListItemAvatar,
   ListItemText,
+  makeStyles,
   Paper,
   Toolbar,
   Typography
 } from "@material-ui/core"
 import { red } from "@material-ui/core/colors"
-import { makeStyles } from "@material-ui/core/styles"
 import ArchiveIcon from "@material-ui/icons/Archive"
 import CheckIcon from "@material-ui/icons/Check"
 import ChevronLeftIcon from "@material-ui/icons/ChevronLeft"
@@ -142,6 +143,13 @@ export default function Dashboard({ accountId, navigate }: Props) {
 
   const [selected, dispatch] = Sel.useSelectedConversations(conversations)
 
+  const searchLoading =
+    searchResult.loading ||
+    (searchResult.data &&
+      searchResult.data.account &&
+      searchResult.data.account.search.loading)
+  const stillLoading = skipSearch ? getAccountResult.loading : searchLoading
+
   //if not all selected star, we want to star instead of unstar
   const isStarred =
     !!conversations &&
@@ -207,11 +215,12 @@ export default function Dashboard({ accountId, navigate }: Props) {
             dispatch={dispatch}
             navigate={navigate}
           />
-        ) : conversations ? (
-          "No conversations to display"
-        ) : (
+        ) : stillLoading ? (
           "Loading..."
+        ) : (
+          "No conversations to display"
         )}
+        {stillLoading ? <LinearProgress /> : null}
       </main>
       <ComposeButton accountId={accountId} />
       <DisplayErrors results={[getAccountResult, searchResult]} />

packages/client/src/documents/operations.graphql

@@ -49,6 +49,7 @@ query searchConversations($accountId: ID!, $query: String!) {
       conversations {
         ...ConversationFieldsForListView
       }
+      loading
       query
     }
   }

packages/client/src/testing/fixtures.ts

@@ -191,7 +191,8 @@ export const conversation2: graphql.Conversation = {
           },
           type: "text",
           subtype: "plain",
-          content: "It's another conversation"
+          content: "It's another conversation",
+          disposition: graphql.Disposition.Inline
         }
       ],
       date: "2019-07-19T12:03:11.114Z",
@@ -224,7 +225,8 @@ export const conversation2: graphql.Conversation = {
           },
           type: "text",
           subtype: "plain",
-          content: "What, again?"
+          content: "What, again?",
+          disposition: graphql.Disposition.Inline
         }
       ],
       date: "2019-07-19T12:21:00.002Z",
@@ -415,7 +417,8 @@ export function searchMock({ query }: { query: string }) {
             __typename: "Search",
             id: 1,
             conversations: [conversation2],
-            query: query
+            loading: false,
+            query
           }
         }
       }

packages/main/package.json

@@ -24,6 +24,7 @@
     "@types/graphql": "^14.2.2",
     "@types/html-to-text": "^1.4.31",
     "@types/kefir": "^3.8.1",
+    "@types/lodash": "^4.14.136",
     "@types/mailparser": "^2.4.0",
     "@types/node": "^11.11.8",
     "@types/node-fetch": "^2.1.7",
@@ -53,6 +54,7 @@
     "kefir": "^3.8.6",
     "keytar": "^4.6.0",
     "libqp": "^1.1.0",
+    "lodash": "^4.17.15",
     "mailparser": "^2.6.0",
     "mkdirp": "^0.5.1",
     "moment": "^2.24.0",

packages/main/schema/schema.graphql

@@ -137,6 +137,7 @@ input PartSpecInput {
 type Search {
   id: ID!
   conversations: [Conversation!]!
+  loading: Boolean!
   query: String!
 }
 

packages/main/src/queue/combineHandlers.ts

@@ -11,7 +11,12 @@ export interface Handler<EnqueueParams, Result, ProcessParams = unknown> {
   enqueue(actionPayload: EnqueueParams): ProcessParams
   process(task: ProcessParams): Promise<Result>
   failure?: (error: Error, task: ProcessParams) => void
+  merge?: (
+    oldTask: ProcessParams,
+    newTask: ProcessParams
+  ) => Promise<ProcessParams>
   priority?: number
+  unique?: boolean
 }
 
 type Handlers = Record<string, Handler<any, any, any>>
@@ -38,7 +43,7 @@ type ActionTypes<HM extends Handlers> = ActionTypeMap<HM>[keyof HM]
 
 type TaskTypeMap<HM extends Handlers> = {
   [K in keyof HM]: HM[K] extends Handler<any, any, infer ProcessParams>
-    ? { type: K; params: ProcessParams }
+    ? { type: K; params: ProcessParams; id?: string }
     : never
 }
 
@@ -55,12 +60,9 @@ export const DEFAULT_PRIORITY = 50
 export const HIGH_PRIORITY = 80
 export const LOW_PRIORITY = 20
 
-export function handler<EnqueueParams, Result, ProcessParams>(h: {
-  enqueue: (params: EnqueueParams) => ProcessParams
-  process: (params: ProcessParams) => Promise<Result>
-  failure?: (error: Error, params: ProcessParams) => void
-  priority?: number // higher numbers run first
-}): Handler<EnqueueParams, Result, ProcessParams> {
+export function handler<EnqueueParams, Result, ProcessParams>(
+  h: Handler<EnqueueParams, Result, ProcessParams>
+): Handler<EnqueueParams, Result, ProcessParams> {
   return h
 }
 
@@ -73,10 +75,45 @@ export function combineHandlers<HM extends Handlers>(
 ): {
   actions: ActionCreators<HM>
   queue: BetterQueue<Task<HM>>
-  schedule: <T>(action: Action<T> & ActionTypes<HM>) => Promise<T>
+  schedule: <T>(action: Action<T> & ActionTypes<HM>) => Promise<T | undefined>
 } {
   const queue = new BetterQueue(process(handlers), {
     ...queueOptions,
+    async filter(task, cb) {
+      try {
+        const unique = task.type && handlers[task.type].unique
+        const store = queueOptions.store
+        if (
+          unique &&
+          task.id &&
+          typeof store === "object" &&
+          "getTask" in store
+        ) {
+          const existing = await promises.lift1(cb => {
+            store.getTask(task.id, cb)
+          })
+          if (existing) {
+            return cb(null, null as any)
+          }
+        }
+        return cb(null, task)
+      } catch (error) {
+        return cb(error, null as any)
+      }
+    },
+    async merge(oldTask, newTask, cb) {
+      try {
+        const mergFunc = oldTask.type && handlers[oldTask.type].merge
+        if (mergFunc && oldTask.params && newTask.params) {
+          const mergedParams = await mergFunc(oldTask.params, newTask.params)
+          cb(null, { ...oldTask, params: mergedParams })
+        } else {
+          cb(null, newTask)
+        }
+      } catch (error) {
+        cb(error, null as any)
+      }
+    },
     priority(task, cb) {
       const priority = handlers[task.type].priority || DEFAULT_PRIORITY
       cb(null, priority)
@@ -93,15 +130,28 @@ export function combineHandlers<HM extends Handlers>(
 function schedule<HM extends Handlers>(
   handlers: HM,
   queue: BetterQueue<{ type: keyof HM; params: unknown }>
-): <T>(action: Action<T> & ActionTypes<HM>) => Promise<T> {
+): <T>(action: Action<T> & ActionTypes<HM>) => Promise<T | undefined> {
   return async action => {
     const type = action.type
     const handler = handlers[type].enqueue
     if (!handler) {
       throw new Error(`No handler for action type, ${type}`)
     }
     const processParams = handler(action.payload)
-    return promises.lift1(cb => queue.push({ type, params: processParams }, cb))
+    const id = processParams.id && { id: processParams.id }
+    return new Promise((resolve, reject) => {
+      queue.push({ type, params: processParams, ...id }, (error, result) => {
+        // If filter rejects the task the error here will be the string,
+        // "input_rejected".
+        if (error === "input_rejected") {
+          return resolve()
+        }
+        if (error) {
+          return reject(error)
+        }
+        return resolve(result)
+      })
+    })
   }
 }
 

packages/main/src/queue/index.ts

@@ -19,7 +19,9 @@
  * - enqueue : Called immediately. Update the local cache to reflect pending
  *             changes, and compute parameters for the next stage. The return
  *             value of the `enqueue` stage is given as an argument to the
- *             `process` and `failure` stages. The return value *must* be
+ *             `process` and `failure` stages. If the return value is an object
+ *             with an `id` property, that value will be used as the task ID for
+ *             purposes of merging duplicate tasks. The return value *must* be
  *             serializable.
  * - process : Called when the task gets to the front of the queue. Make calls
  *             to IMAP and SMTP services. Returns a promise. The resolved value
@@ -32,6 +34,26 @@
  *             `process` stage. This is the place to undo changes to the local
  *             cache to reflect the fact that expected changes did not take
  *             place server-side.
+ *
+ * In addition to the handler stage callbacks, handlers may have these optional
+ * properties:
+ *
+ * - merge    : When two tasks are scheduled with the same ID the merge callback
+ *              will be called to create one unified tasks. `merge` will be
+ *              called with the return values from the `enqueue` stages of each
+ *              task. `merge` returns a promise that resolves to a value to pass
+ *              to the `process` stage. (Set the ID of a task by including an
+ *              `id` property in the return value of the `enqueue` stage.)
+ * - priority : Determines order in which scheduled tasks process. Default is
+ *              `DEFAULT_PRIORITY`, which runs before `LOW_PRIORITY`, and after
+ *              `HIGH_PRIORITY`. Tasks that send data upstream (e.g. setting
+ *              the "Seen" flag on a message) should run *before* tasks that get
+ *              data from upstream (e.g. sync and search).
+ * - unique   : If set to `true` and there is a task that is already queued with
+ *              the same ID then the new task will *not* be scheduled. Note that
+ *              the `enqueue` callback will run regardless! (Set the ID of
+ *              a task by including an `id` property in the return value of the
+ *              `enqueue` stage.)
  */
 
 import * as fs from "fs"
@@ -164,10 +186,11 @@ const handlers = {
 
   search: handler({
     priority: LOW_PRIORITY,
+    unique: true,
     enqueue(searchRecord: cache.Search) {
-      return searchRecord
+      return { id: `search--${searchRecord.query}`, searchRecord }
     },
-    process(searchRecord: cache.Search) {
+    process({ searchRecord }: { searchRecord: cache.Search }) {
       return withConnectionManager(
         String(searchRecord.account_id),
         connectionManager => search(searchRecord, connectionManager)
@@ -251,8 +274,9 @@ const handlers = {
 
   sync: handler({
     priority: LOW_PRIORITY,
+    unique: true,
     enqueue(params: { accountId: ID }) {
-      return params
+      return { ...params, id: "sync" }
     },
     process({ accountId }: { accountId: ID }) {
       return withConnectionManager(accountId, connectionManager =>

packages/main/src/request/index.ts

@@ -159,7 +159,17 @@ export const { actions, perform } = combineHandlers({
     criteria: any[]
   ): R<imap.UID[]> {
     return withBox(connection, box, () =>
-      kefir.fromNodeCallback(cb => connection.search(criteria, cb as any))
+      kefir.fromNodeCallback(cb => {
+        try {
+          // `search` may throw an error instead of relaying an error via the
+          // callback. This happens for example when searching with the
+          // `X-GM-THRID` criteria if the given thread ID is not a string
+          // containing only numeric digits.
+          connection.search(criteria, cb as any)
+        } catch (error) {
+          cb(error, null as any)
+        }
+      })
     )
   }
 })

packages/main/src/resolvers/account.ts

@@ -34,7 +34,7 @@ export const Account: AccountResolvers = {
     const params = { accountId: account.id, boxId: box.id, query }
     const search = cache.getSearch(params) || cache.initSearch(params)
     if (!cache.isSearchFresh(search)) {
-      await schedule(actions.search(search))
+      schedule(actions.search(search))
     }
     return search
   }