mirror of
https://github.com/alexgo-io/gaze-indexer.git
synced 2026-04-21 01:46:44 +08:00
Compare commits
26 Commits
feature/do
...
feature/s3
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4f2dd80546 | ||
|
|
ad26ea0bff | ||
|
|
23b88c7859 | ||
|
|
d8b8ae42fb | ||
|
|
c1362ae328 | ||
|
|
6585da5907 | ||
|
|
41cb5de9c0 | ||
|
|
fc8bad75a5 | ||
|
|
7373944c85 | ||
|
|
89a2e58622 | ||
|
|
9ea9ebdb30 | ||
|
|
5e46a87201 | ||
|
|
e98c3def55 | ||
|
|
a3f902f5d5 | ||
|
|
e8b4f5a2de | ||
|
|
611717706b | ||
|
|
83b38bc67b | ||
|
|
66ea2766a0 | ||
|
|
575c144428 | ||
|
|
c75b62bdf9 | ||
|
|
c9a5c6d217 | ||
|
|
86716c1915 | ||
|
|
c6057d9511 | ||
|
|
d37be5997b | ||
|
|
fcdecd4046 | ||
|
|
5f9cdd5af1 |
2
.github/workflows/release-server.yml
vendored
2
.github/workflows/release-server.yml
vendored
@@ -32,6 +32,6 @@ jobs:
|
||||
uses: ./.github/workflows/reusable-build-and-push-ghcr.yml
|
||||
with:
|
||||
context: .
|
||||
dockerfile: ./docker/Dockerfile
|
||||
dockerfile: Dockerfile
|
||||
image-repo: "ghcr.io/gaze-network/gaze-indexer"
|
||||
image-tag: ${{ needs.prepare.outputs.tag }}
|
||||
|
||||
674
LICENSE
Normal file
674
LICENSE
Normal file
@@ -0,0 +1,674 @@
|
||||
GNU GENERAL PUBLIC LICENSE
|
||||
Version 3, 29 June 2007
|
||||
|
||||
Copyright (C) 2007 Free Software Foundation, Inc. <https://fsf.org/>
|
||||
Everyone is permitted to copy and distribute verbatim copies
|
||||
of this license document, but changing it is not allowed.
|
||||
|
||||
Preamble
|
||||
|
||||
The GNU General Public License is a free, copyleft license for
|
||||
software and other kinds of works.
|
||||
|
||||
The licenses for most software and other practical works are designed
|
||||
to take away your freedom to share and change the works. By contrast,
|
||||
the GNU General Public License is intended to guarantee your freedom to
|
||||
share and change all versions of a program--to make sure it remains free
|
||||
software for all its users. We, the Free Software Foundation, use the
|
||||
GNU General Public License for most of our software; it applies also to
|
||||
any other work released this way by its authors. You can apply it to
|
||||
your programs, too.
|
||||
|
||||
When we speak of free software, we are referring to freedom, not
|
||||
price. Our General Public Licenses are designed to make sure that you
|
||||
have the freedom to distribute copies of free software (and charge for
|
||||
them if you wish), that you receive source code or can get it if you
|
||||
want it, that you can change the software or use pieces of it in new
|
||||
free programs, and that you know you can do these things.
|
||||
|
||||
To protect your rights, we need to prevent others from denying you
|
||||
these rights or asking you to surrender the rights. Therefore, you have
|
||||
certain responsibilities if you distribute copies of the software, or if
|
||||
you modify it: responsibilities to respect the freedom of others.
|
||||
|
||||
For example, if you distribute copies of such a program, whether
|
||||
gratis or for a fee, you must pass on to the recipients the same
|
||||
freedoms that you received. You must make sure that they, too, receive
|
||||
or can get the source code. And you must show them these terms so they
|
||||
know their rights.
|
||||
|
||||
Developers that use the GNU GPL protect your rights with two steps:
|
||||
(1) assert copyright on the software, and (2) offer you this License
|
||||
giving you legal permission to copy, distribute and/or modify it.
|
||||
|
||||
For the developers' and authors' protection, the GPL clearly explains
|
||||
that there is no warranty for this free software. For both users' and
|
||||
authors' sake, the GPL requires that modified versions be marked as
|
||||
changed, so that their problems will not be attributed erroneously to
|
||||
authors of previous versions.
|
||||
|
||||
Some devices are designed to deny users access to install or run
|
||||
modified versions of the software inside them, although the manufacturer
|
||||
can do so. This is fundamentally incompatible with the aim of
|
||||
protecting users' freedom to change the software. The systematic
|
||||
pattern of such abuse occurs in the area of products for individuals to
|
||||
use, which is precisely where it is most unacceptable. Therefore, we
|
||||
have designed this version of the GPL to prohibit the practice for those
|
||||
products. If such problems arise substantially in other domains, we
|
||||
stand ready to extend this provision to those domains in future versions
|
||||
of the GPL, as needed to protect the freedom of users.
|
||||
|
||||
Finally, every program is threatened constantly by software patents.
|
||||
States should not allow patents to restrict development and use of
|
||||
software on general-purpose computers, but in those that do, we wish to
|
||||
avoid the special danger that patents applied to a free program could
|
||||
make it effectively proprietary. To prevent this, the GPL assures that
|
||||
patents cannot be used to render the program non-free.
|
||||
|
||||
The precise terms and conditions for copying, distribution and
|
||||
modification follow.
|
||||
|
||||
TERMS AND CONDITIONS
|
||||
|
||||
0. Definitions.
|
||||
|
||||
"This License" refers to version 3 of the GNU General Public License.
|
||||
|
||||
"Copyright" also means copyright-like laws that apply to other kinds of
|
||||
works, such as semiconductor masks.
|
||||
|
||||
"The Program" refers to any copyrightable work licensed under this
|
||||
License. Each licensee is addressed as "you". "Licensees" and
|
||||
"recipients" may be individuals or organizations.
|
||||
|
||||
To "modify" a work means to copy from or adapt all or part of the work
|
||||
in a fashion requiring copyright permission, other than the making of an
|
||||
exact copy. The resulting work is called a "modified version" of the
|
||||
earlier work or a work "based on" the earlier work.
|
||||
|
||||
A "covered work" means either the unmodified Program or a work based
|
||||
on the Program.
|
||||
|
||||
To "propagate" a work means to do anything with it that, without
|
||||
permission, would make you directly or secondarily liable for
|
||||
infringement under applicable copyright law, except executing it on a
|
||||
computer or modifying a private copy. Propagation includes copying,
|
||||
distribution (with or without modification), making available to the
|
||||
public, and in some countries other activities as well.
|
||||
|
||||
To "convey" a work means any kind of propagation that enables other
|
||||
parties to make or receive copies. Mere interaction with a user through
|
||||
a computer network, with no transfer of a copy, is not conveying.
|
||||
|
||||
An interactive user interface displays "Appropriate Legal Notices"
|
||||
to the extent that it includes a convenient and prominently visible
|
||||
feature that (1) displays an appropriate copyright notice, and (2)
|
||||
tells the user that there is no warranty for the work (except to the
|
||||
extent that warranties are provided), that licensees may convey the
|
||||
work under this License, and how to view a copy of this License. If
|
||||
the interface presents a list of user commands or options, such as a
|
||||
menu, a prominent item in the list meets this criterion.
|
||||
|
||||
1. Source Code.
|
||||
|
||||
The "source code" for a work means the preferred form of the work
|
||||
for making modifications to it. "Object code" means any non-source
|
||||
form of a work.
|
||||
|
||||
A "Standard Interface" means an interface that either is an official
|
||||
standard defined by a recognized standards body, or, in the case of
|
||||
interfaces specified for a particular programming language, one that
|
||||
is widely used among developers working in that language.
|
||||
|
||||
The "System Libraries" of an executable work include anything, other
|
||||
than the work as a whole, that (a) is included in the normal form of
|
||||
packaging a Major Component, but which is not part of that Major
|
||||
Component, and (b) serves only to enable use of the work with that
|
||||
Major Component, or to implement a Standard Interface for which an
|
||||
implementation is available to the public in source code form. A
|
||||
"Major Component", in this context, means a major essential component
|
||||
(kernel, window system, and so on) of the specific operating system
|
||||
(if any) on which the executable work runs, or a compiler used to
|
||||
produce the work, or an object code interpreter used to run it.
|
||||
|
||||
The "Corresponding Source" for a work in object code form means all
|
||||
the source code needed to generate, install, and (for an executable
|
||||
work) run the object code and to modify the work, including scripts to
|
||||
control those activities. However, it does not include the work's
|
||||
System Libraries, or general-purpose tools or generally available free
|
||||
programs which are used unmodified in performing those activities but
|
||||
which are not part of the work. For example, Corresponding Source
|
||||
includes interface definition files associated with source files for
|
||||
the work, and the source code for shared libraries and dynamically
|
||||
linked subprograms that the work is specifically designed to require,
|
||||
such as by intimate data communication or control flow between those
|
||||
subprograms and other parts of the work.
|
||||
|
||||
The Corresponding Source need not include anything that users
|
||||
can regenerate automatically from other parts of the Corresponding
|
||||
Source.
|
||||
|
||||
The Corresponding Source for a work in source code form is that
|
||||
same work.
|
||||
|
||||
2. Basic Permissions.
|
||||
|
||||
All rights granted under this License are granted for the term of
|
||||
copyright on the Program, and are irrevocable provided the stated
|
||||
conditions are met. This License explicitly affirms your unlimited
|
||||
permission to run the unmodified Program. The output from running a
|
||||
covered work is covered by this License only if the output, given its
|
||||
content, constitutes a covered work. This License acknowledges your
|
||||
rights of fair use or other equivalent, as provided by copyright law.
|
||||
|
||||
You may make, run and propagate covered works that you do not
|
||||
convey, without conditions so long as your license otherwise remains
|
||||
in force. You may convey covered works to others for the sole purpose
|
||||
of having them make modifications exclusively for you, or provide you
|
||||
with facilities for running those works, provided that you comply with
|
||||
the terms of this License in conveying all material for which you do
|
||||
not control copyright. Those thus making or running the covered works
|
||||
for you must do so exclusively on your behalf, under your direction
|
||||
and control, on terms that prohibit them from making any copies of
|
||||
your copyrighted material outside their relationship with you.
|
||||
|
||||
Conveying under any other circumstances is permitted solely under
|
||||
the conditions stated below. Sublicensing is not allowed; section 10
|
||||
makes it unnecessary.
|
||||
|
||||
3. Protecting Users' Legal Rights From Anti-Circumvention Law.
|
||||
|
||||
No covered work shall be deemed part of an effective technological
|
||||
measure under any applicable law fulfilling obligations under article
|
||||
11 of the WIPO copyright treaty adopted on 20 December 1996, or
|
||||
similar laws prohibiting or restricting circumvention of such
|
||||
measures.
|
||||
|
||||
When you convey a covered work, you waive any legal power to forbid
|
||||
circumvention of technological measures to the extent such circumvention
|
||||
is effected by exercising rights under this License with respect to
|
||||
the covered work, and you disclaim any intention to limit operation or
|
||||
modification of the work as a means of enforcing, against the work's
|
||||
users, your or third parties' legal rights to forbid circumvention of
|
||||
technological measures.
|
||||
|
||||
4. Conveying Verbatim Copies.
|
||||
|
||||
You may convey verbatim copies of the Program's source code as you
|
||||
receive it, in any medium, provided that you conspicuously and
|
||||
appropriately publish on each copy an appropriate copyright notice;
|
||||
keep intact all notices stating that this License and any
|
||||
non-permissive terms added in accord with section 7 apply to the code;
|
||||
keep intact all notices of the absence of any warranty; and give all
|
||||
recipients a copy of this License along with the Program.
|
||||
|
||||
You may charge any price or no price for each copy that you convey,
|
||||
and you may offer support or warranty protection for a fee.
|
||||
|
||||
5. Conveying Modified Source Versions.
|
||||
|
||||
You may convey a work based on the Program, or the modifications to
|
||||
produce it from the Program, in the form of source code under the
|
||||
terms of section 4, provided that you also meet all of these conditions:
|
||||
|
||||
a) The work must carry prominent notices stating that you modified
|
||||
it, and giving a relevant date.
|
||||
|
||||
b) The work must carry prominent notices stating that it is
|
||||
released under this License and any conditions added under section
|
||||
7. This requirement modifies the requirement in section 4 to
|
||||
"keep intact all notices".
|
||||
|
||||
c) You must license the entire work, as a whole, under this
|
||||
License to anyone who comes into possession of a copy. This
|
||||
License will therefore apply, along with any applicable section 7
|
||||
additional terms, to the whole of the work, and all its parts,
|
||||
regardless of how they are packaged. This License gives no
|
||||
permission to license the work in any other way, but it does not
|
||||
invalidate such permission if you have separately received it.
|
||||
|
||||
d) If the work has interactive user interfaces, each must display
|
||||
Appropriate Legal Notices; however, if the Program has interactive
|
||||
interfaces that do not display Appropriate Legal Notices, your
|
||||
work need not make them do so.
|
||||
|
||||
A compilation of a covered work with other separate and independent
|
||||
works, which are not by their nature extensions of the covered work,
|
||||
and which are not combined with it such as to form a larger program,
|
||||
in or on a volume of a storage or distribution medium, is called an
|
||||
"aggregate" if the compilation and its resulting copyright are not
|
||||
used to limit the access or legal rights of the compilation's users
|
||||
beyond what the individual works permit. Inclusion of a covered work
|
||||
in an aggregate does not cause this License to apply to the other
|
||||
parts of the aggregate.
|
||||
|
||||
6. Conveying Non-Source Forms.
|
||||
|
||||
You may convey a covered work in object code form under the terms
|
||||
of sections 4 and 5, provided that you also convey the
|
||||
machine-readable Corresponding Source under the terms of this License,
|
||||
in one of these ways:
|
||||
|
||||
a) Convey the object code in, or embodied in, a physical product
|
||||
(including a physical distribution medium), accompanied by the
|
||||
Corresponding Source fixed on a durable physical medium
|
||||
customarily used for software interchange.
|
||||
|
||||
b) Convey the object code in, or embodied in, a physical product
|
||||
(including a physical distribution medium), accompanied by a
|
||||
written offer, valid for at least three years and valid for as
|
||||
long as you offer spare parts or customer support for that product
|
||||
model, to give anyone who possesses the object code either (1) a
|
||||
copy of the Corresponding Source for all the software in the
|
||||
product that is covered by this License, on a durable physical
|
||||
medium customarily used for software interchange, for a price no
|
||||
more than your reasonable cost of physically performing this
|
||||
conveying of source, or (2) access to copy the
|
||||
Corresponding Source from a network server at no charge.
|
||||
|
||||
c) Convey individual copies of the object code with a copy of the
|
||||
written offer to provide the Corresponding Source. This
|
||||
alternative is allowed only occasionally and noncommercially, and
|
||||
only if you received the object code with such an offer, in accord
|
||||
with subsection 6b.
|
||||
|
||||
d) Convey the object code by offering access from a designated
|
||||
place (gratis or for a charge), and offer equivalent access to the
|
||||
Corresponding Source in the same way through the same place at no
|
||||
further charge. You need not require recipients to copy the
|
||||
Corresponding Source along with the object code. If the place to
|
||||
copy the object code is a network server, the Corresponding Source
|
||||
may be on a different server (operated by you or a third party)
|
||||
that supports equivalent copying facilities, provided you maintain
|
||||
clear directions next to the object code saying where to find the
|
||||
Corresponding Source. Regardless of what server hosts the
|
||||
Corresponding Source, you remain obligated to ensure that it is
|
||||
available for as long as needed to satisfy these requirements.
|
||||
|
||||
e) Convey the object code using peer-to-peer transmission, provided
|
||||
you inform other peers where the object code and Corresponding
|
||||
Source of the work are being offered to the general public at no
|
||||
charge under subsection 6d.
|
||||
|
||||
A separable portion of the object code, whose source code is excluded
|
||||
from the Corresponding Source as a System Library, need not be
|
||||
included in conveying the object code work.
|
||||
|
||||
A "User Product" is either (1) a "consumer product", which means any
|
||||
tangible personal property which is normally used for personal, family,
|
||||
or household purposes, or (2) anything designed or sold for incorporation
|
||||
into a dwelling. In determining whether a product is a consumer product,
|
||||
doubtful cases shall be resolved in favor of coverage. For a particular
|
||||
product received by a particular user, "normally used" refers to a
|
||||
typical or common use of that class of product, regardless of the status
|
||||
of the particular user or of the way in which the particular user
|
||||
actually uses, or expects or is expected to use, the product. A product
|
||||
is a consumer product regardless of whether the product has substantial
|
||||
commercial, industrial or non-consumer uses, unless such uses represent
|
||||
the only significant mode of use of the product.
|
||||
|
||||
"Installation Information" for a User Product means any methods,
|
||||
procedures, authorization keys, or other information required to install
|
||||
and execute modified versions of a covered work in that User Product from
|
||||
a modified version of its Corresponding Source. The information must
|
||||
suffice to ensure that the continued functioning of the modified object
|
||||
code is in no case prevented or interfered with solely because
|
||||
modification has been made.
|
||||
|
||||
If you convey an object code work under this section in, or with, or
|
||||
specifically for use in, a User Product, and the conveying occurs as
|
||||
part of a transaction in which the right of possession and use of the
|
||||
User Product is transferred to the recipient in perpetuity or for a
|
||||
fixed term (regardless of how the transaction is characterized), the
|
||||
Corresponding Source conveyed under this section must be accompanied
|
||||
by the Installation Information. But this requirement does not apply
|
||||
if neither you nor any third party retains the ability to install
|
||||
modified object code on the User Product (for example, the work has
|
||||
been installed in ROM).
|
||||
|
||||
The requirement to provide Installation Information does not include a
|
||||
requirement to continue to provide support service, warranty, or updates
|
||||
for a work that has been modified or installed by the recipient, or for
|
||||
the User Product in which it has been modified or installed. Access to a
|
||||
network may be denied when the modification itself materially and
|
||||
adversely affects the operation of the network or violates the rules and
|
||||
protocols for communication across the network.
|
||||
|
||||
Corresponding Source conveyed, and Installation Information provided,
|
||||
in accord with this section must be in a format that is publicly
|
||||
documented (and with an implementation available to the public in
|
||||
source code form), and must require no special password or key for
|
||||
unpacking, reading or copying.
|
||||
|
||||
7. Additional Terms.
|
||||
|
||||
"Additional permissions" are terms that supplement the terms of this
|
||||
License by making exceptions from one or more of its conditions.
|
||||
Additional permissions that are applicable to the entire Program shall
|
||||
be treated as though they were included in this License, to the extent
|
||||
that they are valid under applicable law. If additional permissions
|
||||
apply only to part of the Program, that part may be used separately
|
||||
under those permissions, but the entire Program remains governed by
|
||||
this License without regard to the additional permissions.
|
||||
|
||||
When you convey a copy of a covered work, you may at your option
|
||||
remove any additional permissions from that copy, or from any part of
|
||||
it. (Additional permissions may be written to require their own
|
||||
removal in certain cases when you modify the work.) You may place
|
||||
additional permissions on material, added by you to a covered work,
|
||||
for which you have or can give appropriate copyright permission.
|
||||
|
||||
Notwithstanding any other provision of this License, for material you
|
||||
add to a covered work, you may (if authorized by the copyright holders of
|
||||
that material) supplement the terms of this License with terms:
|
||||
|
||||
a) Disclaiming warranty or limiting liability differently from the
|
||||
terms of sections 15 and 16 of this License; or
|
||||
|
||||
b) Requiring preservation of specified reasonable legal notices or
|
||||
author attributions in that material or in the Appropriate Legal
|
||||
Notices displayed by works containing it; or
|
||||
|
||||
c) Prohibiting misrepresentation of the origin of that material, or
|
||||
requiring that modified versions of such material be marked in
|
||||
reasonable ways as different from the original version; or
|
||||
|
||||
d) Limiting the use for publicity purposes of names of licensors or
|
||||
authors of the material; or
|
||||
|
||||
e) Declining to grant rights under trademark law for use of some
|
||||
trade names, trademarks, or service marks; or
|
||||
|
||||
f) Requiring indemnification of licensors and authors of that
|
||||
material by anyone who conveys the material (or modified versions of
|
||||
it) with contractual assumptions of liability to the recipient, for
|
||||
any liability that these contractual assumptions directly impose on
|
||||
those licensors and authors.
|
||||
|
||||
All other non-permissive additional terms are considered "further
|
||||
restrictions" within the meaning of section 10. If the Program as you
|
||||
received it, or any part of it, contains a notice stating that it is
|
||||
governed by this License along with a term that is a further
|
||||
restriction, you may remove that term. If a license document contains
|
||||
a further restriction but permits relicensing or conveying under this
|
||||
License, you may add to a covered work material governed by the terms
|
||||
of that license document, provided that the further restriction does
|
||||
not survive such relicensing or conveying.
|
||||
|
||||
If you add terms to a covered work in accord with this section, you
|
||||
must place, in the relevant source files, a statement of the
|
||||
additional terms that apply to those files, or a notice indicating
|
||||
where to find the applicable terms.
|
||||
|
||||
Additional terms, permissive or non-permissive, may be stated in the
|
||||
form of a separately written license, or stated as exceptions;
|
||||
the above requirements apply either way.
|
||||
|
||||
8. Termination.
|
||||
|
||||
You may not propagate or modify a covered work except as expressly
|
||||
provided under this License. Any attempt otherwise to propagate or
|
||||
modify it is void, and will automatically terminate your rights under
|
||||
this License (including any patent licenses granted under the third
|
||||
paragraph of section 11).
|
||||
|
||||
However, if you cease all violation of this License, then your
|
||||
license from a particular copyright holder is reinstated (a)
|
||||
provisionally, unless and until the copyright holder explicitly and
|
||||
finally terminates your license, and (b) permanently, if the copyright
|
||||
holder fails to notify you of the violation by some reasonable means
|
||||
prior to 60 days after the cessation.
|
||||
|
||||
Moreover, your license from a particular copyright holder is
|
||||
reinstated permanently if the copyright holder notifies you of the
|
||||
violation by some reasonable means, this is the first time you have
|
||||
received notice of violation of this License (for any work) from that
|
||||
copyright holder, and you cure the violation prior to 30 days after
|
||||
your receipt of the notice.
|
||||
|
||||
Termination of your rights under this section does not terminate the
|
||||
licenses of parties who have received copies or rights from you under
|
||||
this License. If your rights have been terminated and not permanently
|
||||
reinstated, you do not qualify to receive new licenses for the same
|
||||
material under section 10.
|
||||
|
||||
9. Acceptance Not Required for Having Copies.
|
||||
|
||||
You are not required to accept this License in order to receive or
|
||||
run a copy of the Program. Ancillary propagation of a covered work
|
||||
occurring solely as a consequence of using peer-to-peer transmission
|
||||
to receive a copy likewise does not require acceptance. However,
|
||||
nothing other than this License grants you permission to propagate or
|
||||
modify any covered work. These actions infringe copyright if you do
|
||||
not accept this License. Therefore, by modifying or propagating a
|
||||
covered work, you indicate your acceptance of this License to do so.
|
||||
|
||||
10. Automatic Licensing of Downstream Recipients.
|
||||
|
||||
Each time you convey a covered work, the recipient automatically
|
||||
receives a license from the original licensors, to run, modify and
|
||||
propagate that work, subject to this License. You are not responsible
|
||||
for enforcing compliance by third parties with this License.
|
||||
|
||||
An "entity transaction" is a transaction transferring control of an
|
||||
organization, or substantially all assets of one, or subdividing an
|
||||
organization, or merging organizations. If propagation of a covered
|
||||
work results from an entity transaction, each party to that
|
||||
transaction who receives a copy of the work also receives whatever
|
||||
licenses to the work the party's predecessor in interest had or could
|
||||
give under the previous paragraph, plus a right to possession of the
|
||||
Corresponding Source of the work from the predecessor in interest, if
|
||||
the predecessor has it or can get it with reasonable efforts.
|
||||
|
||||
You may not impose any further restrictions on the exercise of the
|
||||
rights granted or affirmed under this License. For example, you may
|
||||
not impose a license fee, royalty, or other charge for exercise of
|
||||
rights granted under this License, and you may not initiate litigation
|
||||
(including a cross-claim or counterclaim in a lawsuit) alleging that
|
||||
any patent claim is infringed by making, using, selling, offering for
|
||||
sale, or importing the Program or any portion of it.
|
||||
|
||||
11. Patents.
|
||||
|
||||
A "contributor" is a copyright holder who authorizes use under this
|
||||
License of the Program or a work on which the Program is based. The
|
||||
work thus licensed is called the contributor's "contributor version".
|
||||
|
||||
A contributor's "essential patent claims" are all patent claims
|
||||
owned or controlled by the contributor, whether already acquired or
|
||||
hereafter acquired, that would be infringed by some manner, permitted
|
||||
by this License, of making, using, or selling its contributor version,
|
||||
but do not include claims that would be infringed only as a
|
||||
consequence of further modification of the contributor version. For
|
||||
purposes of this definition, "control" includes the right to grant
|
||||
patent sublicenses in a manner consistent with the requirements of
|
||||
this License.
|
||||
|
||||
Each contributor grants you a non-exclusive, worldwide, royalty-free
|
||||
patent license under the contributor's essential patent claims, to
|
||||
make, use, sell, offer for sale, import and otherwise run, modify and
|
||||
propagate the contents of its contributor version.
|
||||
|
||||
In the following three paragraphs, a "patent license" is any express
|
||||
agreement or commitment, however denominated, not to enforce a patent
|
||||
(such as an express permission to practice a patent or covenant not to
|
||||
sue for patent infringement). To "grant" such a patent license to a
|
||||
party means to make such an agreement or commitment not to enforce a
|
||||
patent against the party.
|
||||
|
||||
If you convey a covered work, knowingly relying on a patent license,
|
||||
and the Corresponding Source of the work is not available for anyone
|
||||
to copy, free of charge and under the terms of this License, through a
|
||||
publicly available network server or other readily accessible means,
|
||||
then you must either (1) cause the Corresponding Source to be so
|
||||
available, or (2) arrange to deprive yourself of the benefit of the
|
||||
patent license for this particular work, or (3) arrange, in a manner
|
||||
consistent with the requirements of this License, to extend the patent
|
||||
license to downstream recipients. "Knowingly relying" means you have
|
||||
actual knowledge that, but for the patent license, your conveying the
|
||||
covered work in a country, or your recipient's use of the covered work
|
||||
in a country, would infringe one or more identifiable patents in that
|
||||
country that you have reason to believe are valid.
|
||||
|
||||
If, pursuant to or in connection with a single transaction or
|
||||
arrangement, you convey, or propagate by procuring conveyance of, a
|
||||
covered work, and grant a patent license to some of the parties
|
||||
receiving the covered work authorizing them to use, propagate, modify
|
||||
or convey a specific copy of the covered work, then the patent license
|
||||
you grant is automatically extended to all recipients of the covered
|
||||
work and works based on it.
|
||||
|
||||
A patent license is "discriminatory" if it does not include within
|
||||
the scope of its coverage, prohibits the exercise of, or is
|
||||
conditioned on the non-exercise of one or more of the rights that are
|
||||
specifically granted under this License. You may not convey a covered
|
||||
work if you are a party to an arrangement with a third party that is
|
||||
in the business of distributing software, under which you make payment
|
||||
to the third party based on the extent of your activity of conveying
|
||||
the work, and under which the third party grants, to any of the
|
||||
parties who would receive the covered work from you, a discriminatory
|
||||
patent license (a) in connection with copies of the covered work
|
||||
conveyed by you (or copies made from those copies), or (b) primarily
|
||||
for and in connection with specific products or compilations that
|
||||
contain the covered work, unless you entered into that arrangement,
|
||||
or that patent license was granted, prior to 28 March 2007.
|
||||
|
||||
Nothing in this License shall be construed as excluding or limiting
|
||||
any implied license or other defenses to infringement that may
|
||||
otherwise be available to you under applicable patent law.
|
||||
|
||||
12. No Surrender of Others' Freedom.
|
||||
|
||||
If conditions are imposed on you (whether by court order, agreement or
|
||||
otherwise) that contradict the conditions of this License, they do not
|
||||
excuse you from the conditions of this License. If you cannot convey a
|
||||
covered work so as to satisfy simultaneously your obligations under this
|
||||
License and any other pertinent obligations, then as a consequence you may
|
||||
not convey it at all. For example, if you agree to terms that obligate you
|
||||
to collect a royalty for further conveying from those to whom you convey
|
||||
the Program, the only way you could satisfy both those terms and this
|
||||
License would be to refrain entirely from conveying the Program.
|
||||
|
||||
13. Use with the GNU Affero General Public License.
|
||||
|
||||
Notwithstanding any other provision of this License, you have
|
||||
permission to link or combine any covered work with a work licensed
|
||||
under version 3 of the GNU Affero General Public License into a single
|
||||
combined work, and to convey the resulting work. The terms of this
|
||||
License will continue to apply to the part which is the covered work,
|
||||
but the special requirements of the GNU Affero General Public License,
|
||||
section 13, concerning interaction through a network will apply to the
|
||||
combination as such.
|
||||
|
||||
14. Revised Versions of this License.
|
||||
|
||||
The Free Software Foundation may publish revised and/or new versions of
|
||||
the GNU General Public License from time to time. Such new versions will
|
||||
be similar in spirit to the present version, but may differ in detail to
|
||||
address new problems or concerns.
|
||||
|
||||
Each version is given a distinguishing version number. If the
|
||||
Program specifies that a certain numbered version of the GNU General
|
||||
Public License "or any later version" applies to it, you have the
|
||||
option of following the terms and conditions either of that numbered
|
||||
version or of any later version published by the Free Software
|
||||
Foundation. If the Program does not specify a version number of the
|
||||
GNU General Public License, you may choose any version ever published
|
||||
by the Free Software Foundation.
|
||||
|
||||
If the Program specifies that a proxy can decide which future
|
||||
versions of the GNU General Public License can be used, that proxy's
|
||||
public statement of acceptance of a version permanently authorizes you
|
||||
to choose that version for the Program.
|
||||
|
||||
Later license versions may give you additional or different
|
||||
permissions. However, no additional obligations are imposed on any
|
||||
author or copyright holder as a result of your choosing to follow a
|
||||
later version.
|
||||
|
||||
15. Disclaimer of Warranty.
|
||||
|
||||
THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY
|
||||
APPLICABLE LAW. EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT
|
||||
HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY
|
||||
OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO,
|
||||
THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
|
||||
PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM
|
||||
IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF
|
||||
ALL NECESSARY SERVICING, REPAIR OR CORRECTION.
|
||||
|
||||
16. Limitation of Liability.
|
||||
|
||||
IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING
|
||||
WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MODIFIES AND/OR CONVEYS
|
||||
THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY
|
||||
GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE
|
||||
USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF
|
||||
DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD
|
||||
PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS),
|
||||
EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF
|
||||
SUCH DAMAGES.
|
||||
|
||||
17. Interpretation of Sections 15 and 16.
|
||||
|
||||
If the disclaimer of warranty and limitation of liability provided
|
||||
above cannot be given local legal effect according to their terms,
|
||||
reviewing courts shall apply local law that most closely approximates
|
||||
an absolute waiver of all civil liability in connection with the
|
||||
Program, unless a warranty or assumption of liability accompanies a
|
||||
copy of the Program in return for a fee.
|
||||
|
||||
END OF TERMS AND CONDITIONS
|
||||
|
||||
How to Apply These Terms to Your New Programs
|
||||
|
||||
If you develop a new program, and you want it to be of the greatest
|
||||
possible use to the public, the best way to achieve this is to make it
|
||||
free software which everyone can redistribute and change under these terms.
|
||||
|
||||
To do so, attach the following notices to the program. It is safest
|
||||
to attach them to the start of each source file to most effectively
|
||||
state the exclusion of warranty; and each file should have at least
|
||||
the "copyright" line and a pointer to where the full notice is found.
|
||||
|
||||
<one line to give the program's name and a brief idea of what it does.>
|
||||
Copyright (C) <year> <name of author>
|
||||
|
||||
This program is free software: you can redistribute it and/or modify
|
||||
it under the terms of the GNU General Public License as published by
|
||||
the Free Software Foundation, either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
This program is distributed in the hope that it will be useful,
|
||||
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
GNU General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU General Public License
|
||||
along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
Also add information on how to contact you by electronic and paper mail.
|
||||
|
||||
If the program does terminal interaction, make it output a short
|
||||
notice like this when it starts in an interactive mode:
|
||||
|
||||
<program> Copyright (C) <year> <name of author>
|
||||
This program comes with ABSOLUTELY NO WARRANTY; for details type `show w'.
|
||||
This is free software, and you are welcome to redistribute it
|
||||
under certain conditions; type `show c' for details.
|
||||
|
||||
The hypothetical commands `show w' and `show c' should show the appropriate
|
||||
parts of the General Public License. Of course, your program's commands
|
||||
might be different; for a GUI interface, you would use an "about box".
|
||||
|
||||
You should also get your employer (if you work as a programmer) or school,
|
||||
if any, to sign a "copyright disclaimer" for the program, if necessary.
|
||||
For more information on this, and how to apply and follow the GNU GPL, see
|
||||
<https://www.gnu.org/licenses/>.
|
||||
|
||||
The GNU General Public License does not permit incorporating your program
|
||||
into proprietary programs. If your program is a subroutine library, you
|
||||
may consider it more useful to permit linking proprietary applications with
|
||||
the library. If this is what you want to do, use the GNU Lesser General
|
||||
Public License instead of this License. But first, please read
|
||||
<https://www.gnu.org/licenses/why-not-lgpl.html>.
|
||||
104
README.md
104
README.md
@@ -1,17 +1,16 @@
|
||||
<!-- omit from toc -->
|
||||
|
||||
# Gaze Indexer
|
||||
Gaze Indexer is an open-source and modular indexing client for Bitcoin meta-protocols. It has support for Bitcoin and Runes out of the box, with **Unified Consistent APIs** across fungible token protocols.
|
||||
|
||||
Gaze Indexer is an open-source and modular indexing client for Bitcoin meta-protocols. It has support for Runes out of the box, with **Unified Consistent APIs** across fungible token protocols.
|
||||
|
||||
Gaze Indexer is built with **modularity** in mind, allowing users to run all modules in one monolithic instance with a single command, or as a distributed cluster of micro-services.
|
||||
|
||||
Gaze Indexer serves as a foundation for building ANY meta-protocol indexers, with efficient data fetching, reorg detection, and database migration tool.
|
||||
Gaze Indexer serves as a foundation for building ANY meta-protocol indexers, with efficient data fetching, reorg detection, and database migration tool.
|
||||
This allows developers to focus on what **truly** matters: Meta-protocol indexing logic. New meta-protocols can be easily added by implementing new modules.
|
||||
|
||||
Gaze Indexer also comes with a block reporting system for verifying data integrity of indexers. Visit the [Gaze Network dashboard](https://dash.gaze.network) to see the status of other indexers.
|
||||
|
||||
- [Modules](#modules)
|
||||
- [1. Bitcoin](#1-bitcoin)
|
||||
- [2. Runes](#2-runes)
|
||||
- [1. Runes](#1-runes)
|
||||
- [Installation](#installation)
|
||||
- [Prerequisites](#prerequisites)
|
||||
- [1. Hardware Requirements](#1-hardware-requirements)
|
||||
@@ -22,73 +21,74 @@ Gaze Indexer also comes with a block reporting system for verifying data integri
|
||||
- [Install from source](#install-from-source)
|
||||
|
||||
## Modules
|
||||
### 1. Bitcoin
|
||||
The Bitcoin Indexer, the heart of every meta-protocol, is responsible for indexing **Bitcoin transactions, blocks, and UTXOs**. It requires a Bitcoin Core RPC as source of Bitcoin transactions,
|
||||
and stores the indexed data in database to be used by other modules.
|
||||
|
||||
### 2. Runes
|
||||
### 1. Runes
|
||||
|
||||
The Runes Indexer is our first meta-protocol indexer. It indexes Runes states, transactions, runestones, and balances using Bitcoin transactions.
|
||||
It comes with a set of APIs for querying historical Runes data. See our [API Reference](https://documenter.getpostman.com/view/28396285/2sA3Bn7Cxr) for full details.
|
||||
|
||||
## Installation
|
||||
### Prerequisites
|
||||
#### 1. Hardware Requirements
|
||||
Each module requires different hardware requirements.
|
||||
| Module | CPU | RAM |
|
||||
| ------- | ---------- | ------ |
|
||||
| Bitcoin | 0.25 cores | 256 MB |
|
||||
| Runes | 0.5 cores | 1 GB |
|
||||
|
||||
#### 2. Prepare Bitcoin Core RPC server.
|
||||
### Prerequisites
|
||||
|
||||
#### 1. Hardware Requirements
|
||||
|
||||
Each module requires different hardware requirements.
|
||||
| Module | CPU | RAM |
|
||||
| ------ | --------- | ---- |
|
||||
| Runes | 0.5 cores | 1 GB |
|
||||
|
||||
#### 2. Prepare Bitcoin Core RPC server.
|
||||
|
||||
Gaze Indexer needs to fetch transaction data from a Bitcoin Core RPC, either self-hosted or using managed providers like QuickNode.
|
||||
To self host a Bitcoin Core, see https://bitcoin.org/en/full-node.
|
||||
|
||||
#### 3. Prepare database.
|
||||
|
||||
Gaze Indexer has first-class support for PostgreSQL. If you wish to use other databases, you can implement your own database repository that satisfies each module's Data Gateway interface.
|
||||
Here is our minimum database disk space requirement for each module.
|
||||
| Module | Database Storage |
|
||||
| ------- | ---------------- |
|
||||
| Bitcoin | 240 GB |
|
||||
| Runes | 150 GB |
|
||||
| Module | Database Storage (current) | Database Storage (in 1 year) |
|
||||
| ------ | -------------------------- | ---------------------------- |
|
||||
| Runes | 10 GB | 150 GB |
|
||||
|
||||
Here is our minimum database disk space requirement for each module.
|
||||
|
||||
#### 4. Prepare `config.yaml` file.
|
||||
|
||||
```yaml
|
||||
# config.yaml
|
||||
logger:
|
||||
output: text # Output format for logs. current supported formats: "text" | "json" | "gcp"
|
||||
output: TEXT # Output format for logs. current supported formats: "TEXT" | "JSON" | "GCP"
|
||||
debug: false
|
||||
|
||||
# Network to run the indexer on. Current supported networks: "mainnet" | "testnet"
|
||||
network: mainnet
|
||||
|
||||
# Bitcoin Core RPC configuration options.
|
||||
bitcoin_node:
|
||||
host: "" # [Required] Host of Bitcoin Core RPC (without https://)
|
||||
user: "" # Username to authenticate with Bitcoin Core RPC
|
||||
pass: "" # Password to authenticate with Bitcoin Core RPC
|
||||
disable_tls: false # Set to true to disable tls
|
||||
|
||||
network: mainnet # Network to run the indexer on. Current supported networks: "mainnet" | "testnet"
|
||||
|
||||
reporting: # Block reporting configuration options. See Block Reporting section for more details.
|
||||
# Block reporting configuration options. See Block Reporting section for more details.
|
||||
reporting:
|
||||
disabled: false # Set to true to disable block reporting to Gaze Network. Default is false.
|
||||
base_url: "https://indexer.api.gaze.network" # Defaults to "https://indexer.api.gaze.network" if left empty
|
||||
name: "" # [Required if not disabled] Name of this indexer to show on the Gaze Network dashboard
|
||||
website_url: "" # Public website URL to show on the dashboard. Can be left empty.
|
||||
indexer_api_url: "" # Public url to access this indexer's API. Can be left empty if you want to keep your indexer private.
|
||||
|
||||
# HTTP server configuration options.
|
||||
http_server:
|
||||
port: 8080 # Port to run the HTTP server on for modules with HTTP API handlers.
|
||||
|
||||
# Meta-protocol modules configuration options.
|
||||
modules:
|
||||
bitcoin: # Configuration options for Bitcoin module. Can be removed if not used.
|
||||
database: "postgres" # Database to store bitcoin data. current supported databases: "postgres"
|
||||
postgres:
|
||||
host: "localhost"
|
||||
port: 5432
|
||||
user: "postgres"
|
||||
password: "password"
|
||||
db_name: "postgres"
|
||||
# url: "postgres://postgres:password@localhost:5432/postgres?sslmode=prefer" # [Optional] This will override other database credentials above.
|
||||
runes: # Configuration options for Runes module. Can be removed if not used.
|
||||
# Configuration options for Runes module. Can be removed if not used.
|
||||
runes:
|
||||
database: "postgres" # Database to store Runes data. current supported databases: "postgres"
|
||||
datasource: "database" # Data source to be used for Bitcoin data. current supported data sources: "bitcoin-node" | "database". If "database" is used, it will use the database config in bitcoin module as datasource.
|
||||
datasource: "database" # Data source to be used for Bitcoin data. current supported data sources: "bitcoin-node".
|
||||
api_handlers: # API handlers to enable. current supported handlers: "http"
|
||||
- http
|
||||
postgres:
|
||||
@@ -101,7 +101,9 @@ modules:
|
||||
```
|
||||
|
||||
### Install with Docker (recommended)
|
||||
|
||||
We will be using `docker-compose` for our installation guide. Make sure the `docker-compose.yaml` file is in the same directory as the `config.yaml` file.
|
||||
|
||||
```yaml
|
||||
# docker-compose.yaml
|
||||
services:
|
||||
@@ -109,19 +111,25 @@ services:
|
||||
image: ghcr.io/gaze-network/gaze-indexer:v1.0.0
|
||||
container_name: gaze-indexer
|
||||
restart: unless-stopped
|
||||
ports:
|
||||
- 8080:8080 # Expose HTTP server port to host
|
||||
volumes:
|
||||
- './config.yaml:/app/config.yaml' # mount config.yaml file to the container as "/app/config.yaml"
|
||||
command: ["/app/main", "run", "--bitcoin", "--runes"] # Put module flags after "run" commands to select which modules to run.
|
||||
- "./config.yaml:/app/config.yaml" # mount config.yaml file to the container as "/app/config.yaml"
|
||||
command: ["/app/main", "run", "--runes"] # Put module flags after "run" commands to select which modules to run.
|
||||
```
|
||||
|
||||
### Install from source
|
||||
1. Install `go` version 1.22 or higher. See Go installation guide [here](https://go.dev/doc/install).
|
||||
|
||||
1. Install `go` version 1.22 or higher. See Go installation guide [here](https://go.dev/doc/install).
|
||||
2. Clone this repository.
|
||||
|
||||
```bash
|
||||
git clone https://github.com/gaze-network/gaze-indexer.git
|
||||
cd gaze-indexer
|
||||
```
|
||||
|
||||
3. Build the main binary.
|
||||
|
||||
```bash
|
||||
# Get dependencies
|
||||
go mod download
|
||||
@@ -129,11 +137,21 @@ go mod download
|
||||
# Build the main binary
|
||||
go build -o gaze main.go
|
||||
```
|
||||
4. Run the main binary with the `run` command and module flags.
|
||||
|
||||
4. Run database migrations with the `migrate` command and module flags.
|
||||
|
||||
```bash
|
||||
./gaze run --bitcoin --runes
|
||||
./gaze migrate up --runes --database postgres://postgres:password@localhost:5432/postgres
|
||||
```
|
||||
|
||||
5. Start the indexer with the `run` command and module flags.
|
||||
|
||||
```bash
|
||||
./gaze run --modules runes
|
||||
```
|
||||
|
||||
If `config.yaml` is not located at `./app/config.yaml`, use the `--config` flag to specify the path to the `config.yaml` file.
|
||||
|
||||
```bash
|
||||
./gaze run --bitcoin --runes --config /path/to/config.yaml
|
||||
./gaze run --modules runes --config /path/to/config.yaml
|
||||
```
|
||||
|
||||
353
cmd/cmd_run.go
353
cmd/cmd_run.go
@@ -15,19 +15,10 @@ import (
|
||||
"github.com/btcsuite/btcd/rpcclient"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/datasources"
|
||||
"github.com/gaze-network/indexer-network/core/indexers"
|
||||
"github.com/gaze-network/indexer-network/core/indexer"
|
||||
"github.com/gaze-network/indexer-network/internal/config"
|
||||
"github.com/gaze-network/indexer-network/internal/postgres"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/btcclient"
|
||||
btcdatagateway "github.com/gaze-network/indexer-network/modules/bitcoin/datagateway"
|
||||
btcpostgres "github.com/gaze-network/indexer-network/modules/bitcoin/repository/postgres"
|
||||
"github.com/gaze-network/indexer-network/modules/runes"
|
||||
runesapi "github.com/gaze-network/indexer-network/modules/runes/api"
|
||||
runesdatagateway "github.com/gaze-network/indexer-network/modules/runes/datagateway"
|
||||
runespostgres "github.com/gaze-network/indexer-network/modules/runes/repository/postgres"
|
||||
runesusecase "github.com/gaze-network/indexer-network/modules/runes/usecase"
|
||||
"github.com/gaze-network/indexer-network/pkg/automaxprocs"
|
||||
"github.com/gaze-network/indexer-network/pkg/errorhandler"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger/slogx"
|
||||
@@ -35,59 +26,49 @@ import (
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/gofiber/fiber/v2/middleware/compress"
|
||||
fiberrecover "github.com/gofiber/fiber/v2/middleware/recover"
|
||||
"github.com/samber/do/v2"
|
||||
"github.com/samber/lo"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
const (
|
||||
shutdownTimeout = 60 * time.Second
|
||||
// Register Modules
|
||||
var Modules = do.Package(
|
||||
do.LazyNamed("runes", runes.New),
|
||||
)
|
||||
|
||||
type runCmdOptions struct {
|
||||
APIOnly bool
|
||||
Bitcoin bool
|
||||
Runes bool
|
||||
}
|
||||
|
||||
func NewRunCommand() *cobra.Command {
|
||||
opts := &runCmdOptions{}
|
||||
|
||||
// Create command
|
||||
runCmd := &cobra.Command{
|
||||
Use: "run",
|
||||
Short: "Start indexer-network service",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return runHandler(opts, cmd, args)
|
||||
if err := automaxprocs.Init(); err != nil {
|
||||
logger.Error("Failed to set GOMAXPROCS", slogx.Error(err))
|
||||
}
|
||||
return runHandler(cmd, args)
|
||||
},
|
||||
}
|
||||
|
||||
// TODO: separate flags and bind flags to each module cmd package.
|
||||
|
||||
// Add local flags
|
||||
flags := runCmd.Flags()
|
||||
flags.BoolVar(&opts.APIOnly, "api-only", false, "Run only API server")
|
||||
flags.BoolVar(&opts.Bitcoin, "bitcoin", false, "Enable Bitcoin indexer module")
|
||||
flags.String("bitcoin-db", "postgres", `Database to store bitcoin data. current supported databases: "postgres"`)
|
||||
flags.BoolVar(&opts.Runes, "runes", false, "Enable Runes indexer module")
|
||||
flags.String("runes-db", "postgres", `Database to store runes data. current supported databases: "postgres"`)
|
||||
flags.String("runes-datasource", "bitcoin-node", `Datasource to fetch bitcoin data for processing Meta-Protocol data. current supported datasources: "bitcoin-node" | "database"`)
|
||||
flags.Bool("api-only", false, "Run only API server")
|
||||
flags.String("modules", "", "Enable specific modules to run. E.g. `runes,brc20`")
|
||||
|
||||
// Bind flags to configuration
|
||||
config.BindPFlag("modules.bitcoin.database", flags.Lookup("bitcoin-db"))
|
||||
config.BindPFlag("modules.runes.database", flags.Lookup("runes-db"))
|
||||
config.BindPFlag("modules.runes.datasource", flags.Lookup("runes-datasource"))
|
||||
config.BindPFlag("api_only", flags.Lookup("api-only"))
|
||||
config.BindPFlag("enable_modules", flags.Lookup("modules"))
|
||||
|
||||
return runCmd
|
||||
}
|
||||
|
||||
type HttpHandler interface {
|
||||
Mount(router fiber.Router) error
|
||||
}
|
||||
const (
|
||||
shutdownTimeout = 60 * time.Second
|
||||
)
|
||||
|
||||
func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
|
||||
func runHandler(cmd *cobra.Command, _ []string) error {
|
||||
conf := config.Load()
|
||||
|
||||
// Validate inputs
|
||||
// Validate inputs and configurations
|
||||
{
|
||||
if !conf.Network.IsSupported() {
|
||||
return errors.Wrapf(errs.Unsupported, "%q network is not supported", conf.Network.String())
|
||||
@@ -98,197 +79,57 @@ func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
|
||||
ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
// Initialize worker context to separate worker's lifecycle from main process
|
||||
ctxWorker, stopWorker := context.WithCancel(context.Background())
|
||||
defer stopWorker()
|
||||
injector := do.New(Modules)
|
||||
do.ProvideValue(injector, conf)
|
||||
do.ProvideValue(injector, ctx)
|
||||
|
||||
// Add logger context
|
||||
ctxWorker = logger.WithContext(ctxWorker, slogx.Stringer("network", conf.Network))
|
||||
// Initialize Bitcoin RPC client
|
||||
do.Provide(injector, func(i do.Injector) (*rpcclient.Client, error) {
|
||||
conf := do.MustInvoke[config.Config](i)
|
||||
|
||||
// Initialize Bitcoin Core RPC Client
|
||||
client, err := rpcclient.New(&rpcclient.ConnConfig{
|
||||
Host: conf.BitcoinNode.Host,
|
||||
User: conf.BitcoinNode.User,
|
||||
Pass: conf.BitcoinNode.Pass,
|
||||
DisableTLS: conf.BitcoinNode.DisableTLS,
|
||||
HTTPPostMode: true,
|
||||
}, nil)
|
||||
if err != nil {
|
||||
logger.PanicContext(ctx, "Invalid Bitcoin node configuration", slogx.Error(err))
|
||||
}
|
||||
defer client.Shutdown()
|
||||
|
||||
// Check Bitcoin RPC connection
|
||||
{
|
||||
start := time.Now()
|
||||
logger.InfoContext(ctx, "Connecting to Bitcoin Core RPC Server...", slogx.String("host", conf.BitcoinNode.Host))
|
||||
if err := client.Ping(); err != nil {
|
||||
logger.PanicContext(ctx, "Can't connect to Bitcoin Core RPC Server", slogx.String("host", conf.BitcoinNode.Host), slogx.Error(err))
|
||||
client, err := rpcclient.New(&rpcclient.ConnConfig{
|
||||
Host: conf.BitcoinNode.Host,
|
||||
User: conf.BitcoinNode.User,
|
||||
Pass: conf.BitcoinNode.Pass,
|
||||
DisableTLS: conf.BitcoinNode.DisableTLS,
|
||||
HTTPPostMode: true,
|
||||
}, nil)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "invalid Bitcoin node configuration")
|
||||
}
|
||||
logger.InfoContext(ctx, "Connected to Bitcoin Core RPC Server", slog.Duration("latency", time.Since(start)))
|
||||
}
|
||||
|
||||
// TODO: create module command package.
|
||||
// each module should have its own command package and main package will routing the command to the module command package.
|
||||
// Check Bitcoin RPC connection
|
||||
{
|
||||
start := time.Now()
|
||||
logger.InfoContext(ctx, "Connecting to Bitcoin Core RPC Server...", slogx.String("host", conf.BitcoinNode.Host))
|
||||
if err := client.Ping(); err != nil {
|
||||
return nil, errors.Wrapf(err, "can't connect to Bitcoin Core RPC Server %q", conf.BitcoinNode.Host)
|
||||
}
|
||||
logger.InfoContext(ctx, "Connected to Bitcoin Core RPC Server", slog.Duration("latency", time.Since(start)))
|
||||
}
|
||||
|
||||
// TODO: refactor module name to specific type instead of string?
|
||||
httpHandlers := make(map[string]HttpHandler, 0)
|
||||
return client, nil
|
||||
})
|
||||
|
||||
var reportingClient *reportingclient.ReportingClient
|
||||
if !conf.Reporting.Disabled {
|
||||
reportingClient, err = reportingclient.New(conf.Reporting)
|
||||
// Initialize reporting client
|
||||
do.Provide(injector, func(i do.Injector) (*reportingclient.ReportingClient, error) {
|
||||
conf := do.MustInvoke[config.Config](i)
|
||||
if conf.Reporting.Disabled {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
reportingClient, err := reportingclient.New(conf.Reporting)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.InvalidArgument) {
|
||||
logger.PanicContext(ctx, "Invalid reporting configuration", slogx.Error(err))
|
||||
return nil, errors.Wrap(err, "invalid reporting configuration")
|
||||
}
|
||||
logger.PanicContext(ctx, "Something went wrong, can't create reporting client", slogx.Error(err))
|
||||
return nil, errors.Wrap(err, "can't create reporting client")
|
||||
}
|
||||
}
|
||||
return reportingClient, nil
|
||||
})
|
||||
|
||||
// Initialize Bitcoin Indexer
|
||||
if opts.Bitcoin {
|
||||
ctx := logger.WithContext(ctx, slogx.String("module", "bitcoin"))
|
||||
var (
|
||||
btcDB btcdatagateway.BitcoinDataGateway
|
||||
indexerInfoDB btcdatagateway.IndexerInformationDataGateway
|
||||
)
|
||||
switch strings.ToLower(conf.Modules.Bitcoin.Database) {
|
||||
case "postgresql", "postgres", "pg":
|
||||
pg, err := postgres.NewPool(ctx, conf.Modules.Bitcoin.Postgres)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.InvalidArgument) {
|
||||
logger.PanicContext(ctx, "Invalid Postgres configuration for indexer", slogx.Error(err))
|
||||
}
|
||||
logger.PanicContext(ctx, "Something went wrong, can't create Postgres connection pool", slogx.Error(err))
|
||||
}
|
||||
defer pg.Close()
|
||||
repo := btcpostgres.NewRepository(pg)
|
||||
btcDB = repo
|
||||
indexerInfoDB = repo
|
||||
default:
|
||||
return errors.Wrapf(errs.Unsupported, "%q database for indexer is not supported", conf.Modules.Bitcoin.Database)
|
||||
}
|
||||
if !opts.APIOnly {
|
||||
processor := bitcoin.NewProcessor(conf, btcDB, indexerInfoDB)
|
||||
datasource := datasources.NewBitcoinNode(client)
|
||||
indexer := indexers.NewBitcoinIndexer(processor, datasource)
|
||||
defer func() {
|
||||
if err := indexer.ShutdownWithTimeout(shutdownTimeout); err != nil {
|
||||
logger.ErrorContext(ctx, "Error during shutdown indexer", slogx.Error(err))
|
||||
return
|
||||
}
|
||||
logger.InfoContext(ctx, "Indexer stopped gracefully")
|
||||
}()
|
||||
|
||||
// Verify states before running Indexer
|
||||
if err := processor.VerifyStates(ctx); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
|
||||
// Run Indexer
|
||||
go func() {
|
||||
// stop main process if indexer stopped
|
||||
defer stop()
|
||||
|
||||
logger.InfoContext(ctx, "Starting Gaze Indexer")
|
||||
if err := indexer.Run(ctxWorker); err != nil {
|
||||
logger.PanicContext(ctx, "Something went wrong, error during running indexer", slogx.Error(err))
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// Initialize Runes Indexer
|
||||
if opts.Runes {
|
||||
ctx := logger.WithContext(ctx, slogx.String("module", "runes"))
|
||||
var (
|
||||
runesDg runesdatagateway.RunesDataGateway
|
||||
indexerInfoDg runesdatagateway.IndexerInfoDataGateway
|
||||
)
|
||||
switch strings.ToLower(conf.Modules.Runes.Database) {
|
||||
case "postgresql", "postgres", "pg":
|
||||
pg, err := postgres.NewPool(ctx, conf.Modules.Runes.Postgres)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.InvalidArgument) {
|
||||
logger.PanicContext(ctx, "Invalid Postgres configuration for indexer", slogx.Error(err))
|
||||
}
|
||||
logger.PanicContext(ctx, "Something went wrong, can't create Postgres connection pool", slogx.Error(err))
|
||||
}
|
||||
defer pg.Close()
|
||||
runesRepo := runespostgres.NewRepository(pg)
|
||||
runesDg = runesRepo
|
||||
indexerInfoDg = runesRepo
|
||||
default:
|
||||
return errors.Wrapf(errs.Unsupported, "%q database for indexer is not supported", conf.Modules.Runes.Database)
|
||||
}
|
||||
var bitcoinDatasource indexers.BitcoinDatasource
|
||||
var bitcoinClient btcclient.Contract
|
||||
switch strings.ToLower(conf.Modules.Runes.Datasource) {
|
||||
case "bitcoin-node":
|
||||
bitcoinNodeDatasource := datasources.NewBitcoinNode(client)
|
||||
bitcoinDatasource = bitcoinNodeDatasource
|
||||
bitcoinClient = bitcoinNodeDatasource
|
||||
case "database":
|
||||
pg, err := postgres.NewPool(ctx, conf.Modules.Bitcoin.Postgres)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.InvalidArgument) {
|
||||
logger.PanicContext(ctx, "Invalid Postgres configuration for datasource", slogx.Error(err))
|
||||
}
|
||||
logger.PanicContext(ctx, "Something went wrong, can't create Postgres connection pool", slogx.Error(err))
|
||||
}
|
||||
defer pg.Close()
|
||||
btcRepo := btcpostgres.NewRepository(pg)
|
||||
btcClientDB := btcclient.NewClientDatabase(btcRepo)
|
||||
bitcoinDatasource = btcClientDB
|
||||
bitcoinClient = btcClientDB
|
||||
default:
|
||||
return errors.Wrapf(errs.Unsupported, "%q datasource is not supported", conf.Modules.Runes.Datasource)
|
||||
}
|
||||
|
||||
if !opts.APIOnly {
|
||||
processor := runes.NewProcessor(runesDg, indexerInfoDg, bitcoinClient, bitcoinDatasource, conf.Network, reportingClient)
|
||||
indexer := indexers.NewBitcoinIndexer(processor, bitcoinDatasource)
|
||||
defer func() {
|
||||
if err := indexer.ShutdownWithTimeout(shutdownTimeout); err != nil {
|
||||
logger.ErrorContext(ctx, "Error during shutdown indexer", slogx.Error(err))
|
||||
return
|
||||
}
|
||||
logger.InfoContext(ctx, "Indexer stopped gracefully")
|
||||
}()
|
||||
|
||||
if err := processor.VerifyStates(ctx); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
|
||||
// Run Indexer
|
||||
go func() {
|
||||
// stop main process if indexer stopped
|
||||
defer stop()
|
||||
|
||||
logger.InfoContext(ctx, "Starting Gaze Indexer")
|
||||
if err := indexer.Run(ctxWorker); err != nil {
|
||||
logger.PanicContext(ctx, "Something went wrong, error during running indexer", slogx.Error(err))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Mount API
|
||||
apiHandlers := lo.Uniq(conf.Modules.Runes.APIHandlers)
|
||||
for _, handler := range apiHandlers {
|
||||
switch handler { // TODO: support more handlers (e.g. gRPC)
|
||||
case "http":
|
||||
runesUsecase := runesusecase.New(runesDg, bitcoinClient)
|
||||
runesHTTPHandler := runesapi.NewHTTPHandler(conf.Network, runesUsecase)
|
||||
httpHandlers["runes"] = runesHTTPHandler
|
||||
default:
|
||||
logger.PanicContext(ctx, "Something went wrong, unsupported API handler", slogx.String("handler", handler))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for interrupt signal to gracefully stop the server with
|
||||
// Setup HTTP server if there are any HTTP handlers
|
||||
if len(httpHandlers) > 0 {
|
||||
// Initialize HTTP server
|
||||
do.Provide(injector, func(i do.Injector) (*fiber.App, error) {
|
||||
app := fiber.New(fiber.Config{
|
||||
AppName: "Gaze Indexer",
|
||||
ErrorHandler: errorhandler.NewHTTPErrorHandler(),
|
||||
@@ -306,38 +147,64 @@ func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
|
||||
Level: compress.LevelDefault,
|
||||
}))
|
||||
|
||||
defer func() {
|
||||
if err := app.ShutdownWithTimeout(shutdownTimeout); err != nil {
|
||||
logger.ErrorContext(ctx, "Error during shutdown HTTP server", slogx.Error(err))
|
||||
return
|
||||
}
|
||||
logger.InfoContext(ctx, "HTTP server stopped gracefully")
|
||||
}()
|
||||
|
||||
// Health check
|
||||
app.Get("/", func(c *fiber.Ctx) error {
|
||||
return errors.WithStack(c.SendStatus(http.StatusOK))
|
||||
})
|
||||
|
||||
// mount http handlers from each http-enabled module
|
||||
for module, handler := range httpHandlers {
|
||||
if err := handler.Mount(app); err != nil {
|
||||
logger.PanicContext(ctx, "Something went wrong, can't mount HTTP handler", slogx.Error(err), slogx.String("module", module))
|
||||
return app, nil
|
||||
})
|
||||
|
||||
// Initialize worker context to separate worker's lifecycle from main process
|
||||
ctxWorker, stopWorker := context.WithCancel(context.Background())
|
||||
defer stopWorker()
|
||||
|
||||
// Add logger context
|
||||
ctxWorker = logger.WithContext(ctxWorker, slogx.Stringer("network", conf.Network))
|
||||
|
||||
// Run modules
|
||||
{
|
||||
modules := lo.Uniq(conf.EnableModules)
|
||||
modules = lo.Map(modules, func(item string, _ int) string { return strings.TrimSpace(item) })
|
||||
modules = lo.Filter(modules, func(item string, _ int) bool { return item != "" })
|
||||
for _, module := range modules {
|
||||
ctx := logger.WithContext(ctxWorker, slogx.String("module", module))
|
||||
|
||||
indexer, err := do.InvokeNamed[indexer.IndexerWorker](injector, module)
|
||||
if err != nil {
|
||||
if errors.Is(err, do.ErrServiceNotFound) {
|
||||
return errors.Errorf("Module %q is not supported", module)
|
||||
}
|
||||
return errors.Wrapf(err, "can't init module %q", module)
|
||||
}
|
||||
|
||||
// Run Indexer
|
||||
if !conf.APIOnly {
|
||||
go func() {
|
||||
// stop main process if indexer stopped
|
||||
defer stop()
|
||||
|
||||
logger.InfoContext(ctx, "Starting Gaze Indexer")
|
||||
if err := indexer.Run(ctx); err != nil {
|
||||
logger.PanicContext(ctx, "Something went wrong, error during running indexer", slogx.Error(err))
|
||||
}
|
||||
}()
|
||||
}
|
||||
logger.InfoContext(ctx, "Mounted HTTP handler", slogx.String("module", module))
|
||||
}
|
||||
|
||||
go func() {
|
||||
// stop main process if API stopped
|
||||
defer stop()
|
||||
|
||||
logger.InfoContext(ctx, "Started HTTP server", slog.Int("port", conf.HTTPServer.Port))
|
||||
if err := app.Listen(fmt.Sprintf(":%d", conf.HTTPServer.Port)); err != nil {
|
||||
logger.PanicContext(ctx, "Something went wrong, error during running HTTP server", slogx.Error(err))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Run API server
|
||||
httpServer := do.MustInvoke[*fiber.App](injector)
|
||||
go func() {
|
||||
// stop main process if API stopped
|
||||
defer stop()
|
||||
|
||||
logger.InfoContext(ctx, "Started HTTP server", slog.Int("port", conf.HTTPServer.Port))
|
||||
if err := httpServer.Listen(fmt.Sprintf(":%d", conf.HTTPServer.Port)); err != nil {
|
||||
logger.PanicContext(ctx, "Something went wrong, error during running HTTP server", slogx.Error(err))
|
||||
}
|
||||
}()
|
||||
|
||||
// Stop application if worker context is done
|
||||
go func() {
|
||||
<-ctxWorker.Done()
|
||||
@@ -366,5 +233,9 @@ func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
|
||||
}
|
||||
}()
|
||||
|
||||
if err := injector.Shutdown(); err != nil {
|
||||
logger.PanicContext(ctx, "Failed while gracefully shutting down", slogx.Error(err))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -6,15 +6,13 @@ import (
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/constants"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin"
|
||||
"github.com/gaze-network/indexer-network/modules/runes"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
var versions = map[string]string{
|
||||
"": constants.Version,
|
||||
"bitcoin": bitcoin.Version,
|
||||
"runes": runes.Version,
|
||||
"": constants.Version,
|
||||
"runes": runes.Version,
|
||||
}
|
||||
|
||||
type versionCmdOptions struct {
|
||||
@@ -33,7 +31,7 @@ func NewVersionCommand() *cobra.Command {
|
||||
}
|
||||
|
||||
flags := cmd.Flags()
|
||||
flags.StringVar(&opts.Modules, "module", "", `Show version of a specific module. E.g. "bitcoin" | "runes"`)
|
||||
flags.StringVar(&opts.Modules, "module", "", `Show version of a specific module. E.g. "runes"`)
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"strings"
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
_ "github.com/golang-migrate/migrate/v4/database/postgres"
|
||||
_ "github.com/golang-migrate/migrate/v4/source/file"
|
||||
@@ -16,7 +17,6 @@ import (
|
||||
|
||||
type migrateDownCmdOptions struct {
|
||||
DatabaseURL string
|
||||
Bitcoin bool
|
||||
Runes bool
|
||||
All bool
|
||||
}
|
||||
@@ -59,7 +59,6 @@ func NewMigrateDownCommand() *cobra.Command {
|
||||
}
|
||||
|
||||
flags := cmd.Flags()
|
||||
flags.BoolVar(&opts.Bitcoin, "bitcoin", false, "Apply Bitcoin down migrations")
|
||||
flags.BoolVar(&opts.Runes, "runes", false, "Apply Runes down migrations")
|
||||
flags.StringVar(&opts.DatabaseURL, "database", "", "Database url to run migration on")
|
||||
flags.BoolVar(&opts.All, "all", false, "Confirm apply ALL down migrations without prompt")
|
||||
@@ -92,12 +91,15 @@ func migrateDownHandler(opts *migrateDownCmdOptions, _ *cobra.Command, args migr
|
||||
newDatabaseURL := cloneURLWithQuery(databaseURL, url.Values{"x-migrations-table": {migrationTable}})
|
||||
sourceURL := "file://" + sourcePath
|
||||
m, err := migrate.New(sourceURL, newDatabaseURL.String())
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "no such file or directory") {
|
||||
return errors.Wrap(errs.InternalError, "migrations directory not found")
|
||||
}
|
||||
return errors.Wrap(err, "failed to open database")
|
||||
}
|
||||
m.Log = &consoleLogger{
|
||||
prefix: fmt.Sprintf("[%s] ", module),
|
||||
}
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create Migrate instance")
|
||||
}
|
||||
if args.N == 0 {
|
||||
m.Log.Printf("Applying down migrations...\n")
|
||||
err = m.Down()
|
||||
@@ -114,11 +116,6 @@ func migrateDownHandler(opts *migrateDownCmdOptions, _ *cobra.Command, args migr
|
||||
return nil
|
||||
}
|
||||
|
||||
if opts.Bitcoin {
|
||||
if err := applyDownMigrations("Bitcoin", bitcoinMigrationSource, "bitcoin_schema_migrations"); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
}
|
||||
if opts.Runes {
|
||||
if err := applyDownMigrations("Runes", runesMigrationSource, "runes_schema_migrations"); err != nil {
|
||||
return errors.WithStack(err)
|
||||
|
||||
@@ -4,8 +4,10 @@ import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
_ "github.com/golang-migrate/migrate/v4/database/postgres"
|
||||
_ "github.com/golang-migrate/migrate/v4/source/file"
|
||||
@@ -14,7 +16,6 @@ import (
|
||||
|
||||
type migrateUpCmdOptions struct {
|
||||
DatabaseURL string
|
||||
Bitcoin bool
|
||||
Runes bool
|
||||
}
|
||||
|
||||
@@ -53,7 +54,6 @@ func NewMigrateUpCommand() *cobra.Command {
|
||||
}
|
||||
|
||||
flags := cmd.Flags()
|
||||
flags.BoolVar(&opts.Bitcoin, "bitcoin", false, "Apply Bitcoin up migrations")
|
||||
flags.BoolVar(&opts.Runes, "runes", false, "Apply Runes up migrations")
|
||||
flags.StringVar(&opts.DatabaseURL, "database", "", "Database url to run migration on")
|
||||
|
||||
@@ -76,12 +76,15 @@ func migrateUpHandler(opts *migrateUpCmdOptions, _ *cobra.Command, args migrateU
|
||||
newDatabaseURL := cloneURLWithQuery(databaseURL, url.Values{"x-migrations-table": {migrationTable}})
|
||||
sourceURL := "file://" + sourcePath
|
||||
m, err := migrate.New(sourceURL, newDatabaseURL.String())
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "no such file or directory") {
|
||||
return errors.Wrap(errs.InternalError, "migrations directory not found")
|
||||
}
|
||||
return errors.Wrap(err, "failed to open database")
|
||||
}
|
||||
m.Log = &consoleLogger{
|
||||
prefix: fmt.Sprintf("[%s] ", module),
|
||||
}
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create Migrate instance")
|
||||
}
|
||||
if args.N == 0 {
|
||||
m.Log.Printf("Applying up migrations...\n")
|
||||
err = m.Up()
|
||||
@@ -98,11 +101,6 @@ func migrateUpHandler(opts *migrateUpCmdOptions, _ *cobra.Command, args migrateU
|
||||
return nil
|
||||
}
|
||||
|
||||
if opts.Bitcoin {
|
||||
if err := applyUpMigrations("Bitcoin", bitcoinMigrationSource, "bitcoin_schema_migrations"); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
}
|
||||
if opts.Runes {
|
||||
if err := applyUpMigrations("Runes", runesMigrationSource, "runes_schema_migrations"); err != nil {
|
||||
return errors.WithStack(err)
|
||||
|
||||
@@ -3,8 +3,7 @@ package migrate
|
||||
import "net/url"
|
||||
|
||||
const (
|
||||
bitcoinMigrationSource = "modules/bitcoin/database/postgresql/migrations"
|
||||
runesMigrationSource = "modules/runes/database/postgresql/migrations"
|
||||
runesMigrationSource = "modules/runes/database/postgresql/migrations"
|
||||
)
|
||||
|
||||
func cloneURLWithQuery(u *url.URL, newQuery url.Values) *url.URL {
|
||||
|
||||
@@ -1,39 +1,36 @@
|
||||
logger:
|
||||
output: text
|
||||
output: TEXT # Output format for logs. current supported formats: "TEXT" | "JSON" | "GCP"
|
||||
debug: false
|
||||
|
||||
bitcoin_node:
|
||||
host: "bitcoin-mainnet-archive.allthatnode.com"
|
||||
user: "user"
|
||||
pass: "pass"
|
||||
disable_tls: false
|
||||
|
||||
# Network to run the indexer on. Current supported networks: "mainnet" | "testnet"
|
||||
network: mainnet
|
||||
|
||||
# Bitcoin Core RPC configuration options.
|
||||
bitcoin_node:
|
||||
host: "" # [Required] Host of Bitcoin Core RPC (without https://)
|
||||
user: "" # Username to authenticate with Bitcoin Core RPC
|
||||
pass: "" # Password to authenticate with Bitcoin Core RPC
|
||||
disable_tls: false # Set to true to disable tls
|
||||
|
||||
# Block reporting configuration options. See Block Reporting section for more details.
|
||||
reporting:
|
||||
disabled: false
|
||||
base_url: "https://indexer.api.gaze.network" # defaults to "https://indexer.api.gaze.network" if empty
|
||||
name: "local-dev" # name of this instance to show on the dashboard
|
||||
website_url: "" # public website URL to show on the dashboard. Can be left empty.
|
||||
indexer_api_url: "" # public url to access this api. Can be left empty.
|
||||
disabled: false # Set to true to disable block reporting to Gaze Network. Default is false.
|
||||
base_url: "https://indexer.api.gaze.network" # Defaults to "https://indexer.api.gaze.network" if left empty
|
||||
name: "" # [Required if not disabled] Name of this indexer to show on the Gaze Network dashboard
|
||||
website_url: "" # Public website URL to show on the dashboard. Can be left empty.
|
||||
indexer_api_url: "" # Public url to access this indexer's API. Can be left empty if you want to keep your indexer private.
|
||||
|
||||
้http_server:
|
||||
port: 8080
|
||||
# HTTP server configuration options.
|
||||
http_server:
|
||||
port: 8080 # Port to run the HTTP server on for modules with HTTP API handlers.
|
||||
|
||||
# Meta-protocol modules configuration options.
|
||||
modules:
|
||||
bitcoin:
|
||||
database: "postgres" # Database to store bitcoin data. current supported databases: "postgres"
|
||||
postgres:
|
||||
host: "localhost"
|
||||
port: 5432
|
||||
user: "postgres"
|
||||
password: "password"
|
||||
db_name: "postgres"
|
||||
# url: "postgres://postgres:password@localhost:5432/postgres?sslmode=prefer" # [Optional] This will override other database connection configurations
|
||||
# Configuration options for Runes module. Can be removed if not used.
|
||||
runes:
|
||||
database: "postgres" # Database to store Runes data. current supported databases: "postgres"
|
||||
datasource: "postgres" # Data source to be used (to fetch bitcoin blocks). current supported data sources: "bitcoin-node" | "postgres"
|
||||
api_handlers: # API handlers to be used. current supported handlers: "http"
|
||||
datasource: "database" # Data source to be used for Bitcoin data. current supported data sources: "bitcoin-node".
|
||||
api_handlers: # API handlers to enable. current supported handlers: "http"
|
||||
- http
|
||||
postgres:
|
||||
host: "localhost"
|
||||
@@ -41,4 +38,4 @@ modules:
|
||||
user: "postgres"
|
||||
password: "password"
|
||||
db_name: "postgres"
|
||||
# url: "postgres://postgres:password@localhost:5432/postgres?sslmode=prefer" # [Optional] This will override other database connection configurations
|
||||
# url: "postgres://postgres:password@localhost:5432/postgres?sslmode=prefer" # [Optional] This will override other database credentials above.
|
||||
|
||||
688
core/datasources/aws_public_data.go
Normal file
688
core/datasources/aws_public_data.go
Normal file
@@ -0,0 +1,688 @@
|
||||
// AWS Public Blockchain Datasource
|
||||
// - https://registry.opendata.aws/aws-public-blockchain
|
||||
// - https://github.com/aws-solutions-library-samples/guidance-for-digital-assets-on-aws
|
||||
//
|
||||
// To setup your own data source, see: https://github.com/aws-solutions-library-samples/guidance-for-digital-assets-on-aws/blob/main/analytics/producer/README.md
|
||||
package datasources
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"math"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/btcsuite/btcd/wire"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/internal/subscription"
|
||||
"github.com/gaze-network/indexer-network/pkg/btcutils"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger/slogx"
|
||||
"github.com/gaze-network/indexer-network/pkg/parquetutils"
|
||||
"github.com/samber/lo"
|
||||
parquettypes "github.com/xitongsys/parquet-go/types"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
const (
|
||||
awsPublicDataS3Region = "us-east-2"
|
||||
awsPublicDataS3Bucket = "aws-public-blockchain"
|
||||
|
||||
defaultAWSPublicDataDownloadConcurrency = 8
|
||||
)
|
||||
|
||||
var firstBitcoinTimestamp = time.Date(2009, time.January, 3, 18, 15, 5, 0, time.UTC)
|
||||
|
||||
// Make sure to implement the BitcoinDatasource interface
|
||||
var _ Datasource[*types.Block] = (*AWSPublicDataDatasource)(nil)
|
||||
|
||||
type AWSPublicDataDatasourceConfig struct {
|
||||
// The number of goroutines to spin up in parallel when downloading parts.
|
||||
// Concurrency of 1 will download the parts sequentially.
|
||||
// Default is 8.
|
||||
//
|
||||
// CAUTION: High concurrency with low part size can reduce the time to download file,
|
||||
// but it can also increase the memory usage.
|
||||
DownloadConcurrency int
|
||||
|
||||
// The size (in bytes) to request from S3 for each part.
|
||||
// Default is depend on the concurrency and file size (part size = file size / concurrency).
|
||||
//
|
||||
// CAUTION: High concurrency with low part size can reduce the time to download file,
|
||||
// but it can also increase the memory usage.
|
||||
DownloadPartSize int64 `mapstructure:"download_part_size"`
|
||||
}
|
||||
|
||||
type AWSPublicDataDatasource struct {
|
||||
btcDatasource Datasource[*types.Block]
|
||||
s3Client *s3.Client
|
||||
s3Bucket string
|
||||
config AWSPublicDataDatasourceConfig
|
||||
}
|
||||
|
||||
func NewAWSPublicData(btcDatasource Datasource[*types.Block], conf AWSPublicDataDatasourceConfig) *AWSPublicDataDatasource {
|
||||
sdkConfig, err := config.LoadDefaultConfig(context.Background())
|
||||
if err != nil {
|
||||
logger.Panic("Can't load AWS SDK user config", slogx.Error(err), slog.String("package", "datasources"))
|
||||
}
|
||||
|
||||
// TODO: support user defined config (self-hosted s3 bucket)
|
||||
s3client := s3.NewFromConfig(sdkConfig, func(o *s3.Options) {
|
||||
o.Region = awsPublicDataS3Region
|
||||
o.Credentials = aws.AnonymousCredentials{}
|
||||
})
|
||||
|
||||
if conf.DownloadConcurrency <= 0 {
|
||||
conf.DownloadConcurrency = defaultAWSPublicDataDownloadConcurrency
|
||||
}
|
||||
|
||||
return &AWSPublicDataDatasource{
|
||||
btcDatasource: btcDatasource,
|
||||
s3Client: s3client,
|
||||
s3Bucket: awsPublicDataS3Bucket,
|
||||
config: conf,
|
||||
}
|
||||
}
|
||||
|
||||
func (d AWSPublicDataDatasource) Name() string {
|
||||
return fmt.Sprintf("aws_public_data/%s", d.btcDatasource.Name())
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) Fetch(ctx context.Context, from, to int64) ([]*types.Block, error) {
|
||||
ch := make(chan []*types.Block)
|
||||
subscription, err := d.FetchAsync(ctx, from, to, ch)
|
||||
if err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
}
|
||||
defer subscription.Unsubscribe()
|
||||
|
||||
blocks := make([]*types.Block, 0)
|
||||
for {
|
||||
select {
|
||||
case b, ok := <-ch:
|
||||
if !ok {
|
||||
return blocks, nil
|
||||
}
|
||||
blocks = append(blocks, b...)
|
||||
case <-subscription.Done():
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, errors.Wrap(err, "context done")
|
||||
}
|
||||
return blocks, nil
|
||||
case err := <-subscription.Err():
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "got error while fetch async")
|
||||
}
|
||||
return blocks, nil
|
||||
case <-ctx.Done():
|
||||
return nil, errors.Wrap(ctx.Err(), "context done")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) FetchAsync(ctx context.Context, from, to int64, ch chan<- []*types.Block) (*subscription.ClientSubscription[[]*types.Block], error) {
|
||||
ctx = logger.WithContext(ctx,
|
||||
slogx.String("package", "datasources"),
|
||||
slogx.String("datasource", d.Name()),
|
||||
)
|
||||
|
||||
start, end, skip, err := d.prepareRange(ctx, from, to)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to prepare fetch range")
|
||||
}
|
||||
|
||||
subscription := subscription.NewSubscription(ch)
|
||||
if skip {
|
||||
if err := subscription.UnsubscribeWithContext(ctx); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to unsubscribe")
|
||||
}
|
||||
return subscription.Client(), nil
|
||||
}
|
||||
|
||||
startFiles, err := d.listBlocksFilesByDate(ctx, start.Timestamp)
|
||||
if err != nil {
|
||||
if err := subscription.UnsubscribeWithContext(ctx); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to unsubscribe")
|
||||
}
|
||||
return nil, errors.Wrap(err, "failed to list files by date")
|
||||
}
|
||||
|
||||
// supported only merged blocks files
|
||||
startFiles = lo.Filter(startFiles, func(file awsFile, _ int) bool {
|
||||
return strings.Contains(file.Key, "part-")
|
||||
})
|
||||
|
||||
// use other datasource instead of s3 if there's no supported data
|
||||
if len(startFiles) == 0 {
|
||||
if err := subscription.UnsubscribeWithContext(ctx); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to unsubscribe")
|
||||
}
|
||||
s, err := d.btcDatasource.FetchAsync(ctx, start.Height, end.Height, ch)
|
||||
return s, errors.WithStack(err)
|
||||
}
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
// add a bit delay to prevent shutdown before client receive all blocks
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
subscription.Unsubscribe()
|
||||
}()
|
||||
// loop through each day until reach the end of supported data or within end block date
|
||||
for ts := start.Timestamp; ts.Before(end.Timestamp.Round(24*time.Hour)) && ts.Before(time.Now()); ts = ts.Add(24 * time.Hour) {
|
||||
ctx := logger.WithContext(ctx,
|
||||
slogx.Time("date", ts),
|
||||
slogx.Int64("date_unix", ts.Unix()),
|
||||
)
|
||||
|
||||
logger.DebugContext(ctx, "Fetching data from AWS S3", slogx.Int64("start", start.Height), slogx.Int64("end", end.Height))
|
||||
|
||||
allBlocksFiles, err := d.listBlocksFilesByDate(ctx, ts)
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to list blocks files by date from aws s3", slogx.Error(err))
|
||||
if err := subscription.SendError(ctx, errors.WithStack(err)); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
allTxsFiles, err := d.listTxsFilesByDate(ctx, ts)
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to list txs files by date from aws s3", slogx.Error(err))
|
||||
if err := subscription.SendError(ctx, errors.WithStack(err)); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
blocksFiles := lo.Filter(allBlocksFiles, func(file awsFile, _ int) bool {
|
||||
return strings.Contains(file.Key, "part-")
|
||||
})
|
||||
txsFiles := lo.Filter(allTxsFiles, func(file awsFile, _ int) bool {
|
||||
return strings.Contains(file.Key, "part-")
|
||||
})
|
||||
|
||||
logger.DebugContext(ctx, "Found files in AWS S3 bucket",
|
||||
slogx.Int("files_blocks", len(allBlocksFiles)),
|
||||
slogx.Int("files_blocks_merged", len(blocksFiles)),
|
||||
slogx.Int("files_txs_all", len(allTxsFiles)),
|
||||
slogx.Int("files_txs_merged", len(txsFiles)),
|
||||
)
|
||||
|
||||
// Reach the end of supported data,
|
||||
// stop fetching data from AWS S3
|
||||
if len(blocksFiles) == 0 || len(txsFiles) == 0 {
|
||||
logger.DebugContext(ctx, "No blocks files found, stop fetching data from AWS S3")
|
||||
return
|
||||
}
|
||||
|
||||
// prevent unexpected error
|
||||
{
|
||||
if len(blocksFiles) != 1 {
|
||||
logger.ErrorContext(ctx, "Unexpected blocks files count, should be 1", slogx.Int("count", len(blocksFiles)))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(errs.InternalError, "unexpected blocks files count")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
if len(txsFiles) != 1 {
|
||||
logger.ErrorContext(ctx, "Unexpected txs files count, should be 1", slogx.Int("count", len(txsFiles)))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(errs.InternalError, "unexpected txs files count")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: use concurrent stream (max 2 goroutine) to download files then sequentially read parquet files
|
||||
// to improve performance while not consuming too much memory (increase around 500 MB per goroutine)
|
||||
var (
|
||||
// TODO: create []byte pool to reduce alloc ops (reduce GC pressure)
|
||||
// TODO: use FileSystem for default buffer (can choose memory or disk buffer)
|
||||
blocksBuffer = parquetutils.NewBuffer()
|
||||
txsBuffer = parquetutils.NewBuffer()
|
||||
)
|
||||
startDownload := time.Now()
|
||||
if err := d.downloadFile(ctx, blocksFiles[0], blocksBuffer); err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to download blocks file from AWS S3", slogx.Int("count", len(txsFiles)))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(err, "can't download blocks file")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
}
|
||||
if err := d.downloadFile(ctx, txsFiles[0], txsBuffer); err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to download blocks file from AWS S3", slogx.Int("count", len(txsFiles)))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(err, "can't download blocks file")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
}
|
||||
logger.DebugContext(ctx, "Downloaded files from AWS S3",
|
||||
slogx.Duration("duration", time.Since(startDownload)),
|
||||
slogx.Int("sizes_blocks", len(blocksBuffer.Bytes())),
|
||||
slogx.Int("sizes_txs", len(txsBuffer.Bytes())),
|
||||
)
|
||||
|
||||
// Read parquet files
|
||||
startRead := time.Now()
|
||||
|
||||
// we can read all blocks data at once because it's small
|
||||
rawAllBlocks, err := parquetutils.ReadAll[awsBlock](blocksBuffer)
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to read parquet blocks data", slogx.Error(err))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(err, "can't read parquet blocks data")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// NOTE: We shouldn't read all txs data at once because it's very huge (up to ~1.5GB memory usage)
|
||||
// we should read it by chunk and send it to subscription client to reduce memory usage.
|
||||
// But AWS Public Dataset are not sorted by block number and index,
|
||||
// so we can't avoid reading all transactions data by skip unnecessary transactions
|
||||
// or chunk data by block number to reduce memory usage :(
|
||||
rawAllTxs, err := parquetutils.ReadAll[awsTransaction](blocksBuffer)
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to read parquet txs data", slogx.Error(err))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(err, "can't read parquet blocks data")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
groupRawTxs := lo.GroupBy(rawAllTxs, func(tx awsTransaction) int64 {
|
||||
return tx.BlockNumber
|
||||
})
|
||||
|
||||
// filter blocks data by height range
|
||||
rawFilteredBlocks := lo.Filter(rawAllBlocks, func(block awsBlock, _ int) bool {
|
||||
return block.Number >= start.Height && block.Number <= end.Height
|
||||
})
|
||||
slices.SortFunc(rawFilteredBlocks, func(i, j awsBlock) int {
|
||||
return cmp.Compare(i.Number, j.Number)
|
||||
})
|
||||
|
||||
logger.DebugContext(ctx, "Read parquet files",
|
||||
slogx.Duration("duration", time.Since(startRead)),
|
||||
slogx.Int("total_blocks", len(rawAllBlocks)),
|
||||
slogx.Int("filtered_blocks", len(rawFilteredBlocks)),
|
||||
slogx.Int("total_txs", len(rawAllTxs)),
|
||||
slogx.Int("total_txs_grouped", len(groupRawTxs)),
|
||||
)
|
||||
|
||||
blocks := make([]*types.Block, 0, len(rawFilteredBlocks))
|
||||
for _, rawBlock := range rawFilteredBlocks {
|
||||
blockHeader, err := rawBlock.ToBlockHeader()
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to convert aws block to type block header", slogx.Error(err))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(err, "can't convert aws block to type block header")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
txs := make([]*types.Transaction, 0, len(groupRawTxs[blockHeader.Height]))
|
||||
for _, rawTx := range groupRawTxs[rawBlock.Number] {
|
||||
tx, err := rawTx.ToTransaction(rawBlock)
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Failed to convert aws transaction to type transaction", slogx.Error(err))
|
||||
if err := subscription.SendError(ctx, errors.Wrap(err, "can't convert aws transaction to type transaction")); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
txs = append(txs, tx)
|
||||
}
|
||||
slices.SortFunc(txs, func(i, j *types.Transaction) int {
|
||||
return cmp.Compare(i.Index, j.Index)
|
||||
})
|
||||
|
||||
blocks = append(blocks, &types.Block{
|
||||
Header: blockHeader,
|
||||
Transactions: txs,
|
||||
})
|
||||
}
|
||||
|
||||
logger.DebugContext(ctx, "Send blocks to subscription client", slogx.Int("count", len(blocks)))
|
||||
if err := subscription.Send(ctx, blocks); err != nil {
|
||||
if errors.Is(err, errs.Closed) {
|
||||
logger.DebugContext(ctx, "Subscription client closed, can't send", slogx.Error(err))
|
||||
return
|
||||
}
|
||||
logger.WarnContext(ctx, "Failed to send bitcoin blocks to subscription client",
|
||||
slogx.Int64("start", blocks[0].Header.Height),
|
||||
slogx.Int64("end", blocks[len(blocks)-1].Header.Height),
|
||||
slogx.Error(err),
|
||||
)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return subscription.Client(), nil
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) GetBlockHeader(ctx context.Context, height int64) (types.BlockHeader, error) {
|
||||
header, err := d.btcDatasource.GetBlockHeader(ctx, height)
|
||||
return header, errors.WithStack(err)
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) GetCurrentBlockHeight(ctx context.Context) (int64, error) {
|
||||
height, err := d.btcDatasource.GetCurrentBlockHeight(ctx)
|
||||
return height, errors.WithStack(err)
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) prepareRange(ctx context.Context, fromHeight, toHeight int64) (startHeader, endHeader types.BlockHeader, skip bool, err error) {
|
||||
start := fromHeight
|
||||
end := toHeight
|
||||
|
||||
// get current bitcoin block height
|
||||
latestBlockHeight, err := d.btcDatasource.GetCurrentBlockHeight(ctx)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, types.BlockHeader{}, false, errors.Wrap(err, "failed to get block count")
|
||||
}
|
||||
|
||||
// set start to genesis block height
|
||||
if start < 0 {
|
||||
start = 0
|
||||
}
|
||||
|
||||
// set end to current bitcoin block height if
|
||||
// - end is -1
|
||||
// - end is greater that current bitcoin block height
|
||||
if end < 0 || end > latestBlockHeight {
|
||||
end = latestBlockHeight
|
||||
}
|
||||
|
||||
// if start is greater than end, skip this round
|
||||
if start > end {
|
||||
return types.BlockHeader{}, types.BlockHeader{}, true, nil
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, types.BlockHeader{}, false, errors.Wrapf(err, "block %v", end)
|
||||
}
|
||||
|
||||
group, groupctx := errgroup.WithContext(ctx)
|
||||
group.Go(func() error {
|
||||
startHeader, err = d.GetBlockHeader(groupctx, start)
|
||||
return errors.Wrapf(err, "block %v", start)
|
||||
})
|
||||
group.Go(func() error {
|
||||
endHeader, err = d.GetBlockHeader(ctx, end)
|
||||
return errors.Wrapf(err, "block %v", end)
|
||||
})
|
||||
if err := group.Wait(); err != nil {
|
||||
return types.BlockHeader{}, types.BlockHeader{}, false, errors.Wrap(err, "failed to get block header")
|
||||
}
|
||||
|
||||
return startHeader, endHeader, false, nil
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) listFiles(ctx context.Context, prefix string) ([]awsFile, error) {
|
||||
result, err := d.s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
|
||||
Bucket: aws.String(d.s3Bucket),
|
||||
Prefix: aws.String(prefix),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "can't list s3 bucket objects for bucket %q and prefix %q", d.s3Bucket, prefix)
|
||||
}
|
||||
|
||||
// filter empty keys
|
||||
objs := lo.Filter(result.Contents, func(item s3types.Object, _ int) bool { return item.Key != nil })
|
||||
return lo.Map(objs, func(item s3types.Object, _ int) awsFile {
|
||||
return awsFile{
|
||||
Key: *item.Key,
|
||||
Size: *item.Size,
|
||||
LastModified: *item.LastModified,
|
||||
}
|
||||
}), nil
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) listBlocksFilesByDate(ctx context.Context, date time.Time) ([]awsFile, error) {
|
||||
if date.Before(firstBitcoinTimestamp) {
|
||||
return nil, errors.Wrapf(errs.InvalidArgument, "date %v is before first bitcoin timestamp %v", date, firstBitcoinTimestamp)
|
||||
}
|
||||
prefix := "v1.0/btc/blocks/date=" + date.UTC().Format(time.DateOnly)
|
||||
files, err := d.listFiles(ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to list blocks files by date")
|
||||
}
|
||||
return files, nil
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) listTxsFilesByDate(ctx context.Context, date time.Time) ([]awsFile, error) {
|
||||
if date.Before(firstBitcoinTimestamp) {
|
||||
return nil, errors.Wrapf(errs.InvalidArgument, "date %v is before first bitcoin timestamp %v", date, firstBitcoinTimestamp)
|
||||
}
|
||||
prefix := "v1.0/btc/transactions/date=" + date.UTC().Format(time.DateOnly)
|
||||
files, err := d.listFiles(ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to list txs files by date")
|
||||
}
|
||||
return files, nil
|
||||
}
|
||||
|
||||
func (d *AWSPublicDataDatasource) downloadFile(ctx context.Context, f awsFile, w io.WriterAt) error {
|
||||
downloader := manager.NewDownloader(d.s3Client, func(md *manager.Downloader) {
|
||||
md.Concurrency = d.config.DownloadConcurrency
|
||||
md.PartSize = d.config.DownloadPartSize
|
||||
if md.PartSize <= 0 {
|
||||
md.PartSize = f.Size / int64(md.Concurrency)
|
||||
}
|
||||
})
|
||||
|
||||
numBytes, err := downloader.Download(ctx, w, &s3.GetObjectInput{
|
||||
Bucket: aws.String(d.s3Bucket),
|
||||
Key: aws.String(f.Key),
|
||||
})
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "failed to download file for bucket %q and key %q", d.s3Bucket, f.Key)
|
||||
}
|
||||
|
||||
if numBytes < 1 {
|
||||
return errors.Wrap(errs.NotFound, "got empty file")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// TODO: remove unused fields to reduce memory usage
|
||||
type (
|
||||
awsBlock struct {
|
||||
Hash string `parquet:"name=hash, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
Bits string `parquet:"name=bits, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"` // Hex string format
|
||||
PreviousBlockHash string `parquet:"name=previousblockhash, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
MerkleRoot string `parquet:"name=merkle_root, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
CoinbaseParam string `parquet:"name=coinbase_param, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
Timestamp string `parquet:"name=timestamp, type=INT96, repetitiontype=OPTIONAL"`
|
||||
Number int64 `parquet:"name=number, type=INT64, repetitiontype=OPTIONAL"`
|
||||
Version int64 `parquet:"name=version, type=INT64, repetitiontype=OPTIONAL"`
|
||||
Nonce int64 `parquet:"name=nonce, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// MedianTime string `parquet:"name=mediantime, type=INT96, repetitiontype=OPTIONAL"`
|
||||
// Difficulty float64 `parquet:"name=difficulty, type=DOUBLE, repetitiontype=OPTIONAL"`
|
||||
// Chainwork string `parquet:"name=chainwork, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// Size int64 `parquet:"name=size, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// Weight int64 `parquet:"name=weight, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// TransactionCount int64 `parquet:"name=transaction_count, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// StrippedSize int64 `parquet:"name=stripped_size, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// Date string `parquet:"name=date, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// LastModified string `parquet:"name=last_modified, type=INT96, repetitiontype=OPTIONAL"`
|
||||
}
|
||||
awsTransaction struct {
|
||||
Hash string `parquet:"name=hash, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
BlockHash string `parquet:"name=block_hash, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
Outputs []*awsTxOutput `parquet:"name=outputs, type=LIST, repetitiontype=OPTIONAL, valuetype=STRUCT"`
|
||||
Inputs []*awsTxInput `parquet:"name=inputs, type=LIST, repetitiontype=OPTIONAL, valuetype=STRUCT"`
|
||||
Version int64 `parquet:"name=version, type=INT64, repetitiontype=OPTIONAL"`
|
||||
Size int64 `parquet:"name=size, type=INT64, repetitiontype=OPTIONAL"`
|
||||
BlockNumber int64 `parquet:"name=block_number, type=INT64, repetitiontype=OPTIONAL"`
|
||||
Index int64 `parquet:"name=index, type=INT64, repetitiontype=OPTIONAL"`
|
||||
LockTime int64 `parquet:"name=lock_time, type=INT64, repetitiontype=OPTIONAL"`
|
||||
IsCoinbase bool `parquet:"name=is_coinbase, type=BOOLEAN, repetitiontype=OPTIONAL"`
|
||||
// VirtualSize int64 `parquet:"name=virtual_size, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// InputCount int64 `parquet:"name=input_count, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// OutputCount int64 `parquet:"name=output_count, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// OutputValue float64 `parquet:"name=output_value, type=DOUBLE, repetitiontype=OPTIONAL"`
|
||||
// BlockTimestamp string `parquet:"name=block_timestamp, type=INT96, repetitiontype=OPTIONAL"`
|
||||
// Date string `parquet:"name=date, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// LastModified string `parquet:"name=last_modified, type=INT96, repetitiontype=OPTIONAL"`
|
||||
// Fee float64 `parquet:"name=fee, type=DOUBLE, repetitiontype=OPTIONAL"`
|
||||
// InputValue float64 `parquet:"name=input_value, type=DOUBLE, repetitiontype=OPTIONAL"`
|
||||
}
|
||||
awsTxInput struct {
|
||||
ScriptHex string `parquet:"name=script_hex, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
SpentTransactionHash string `parquet:"name=spent_transaction_hash, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
TxInWitness []*string `parquet:"name=txinwitness, type=LIST, repetitiontype=OPTIONAL, valuetype=BYTE_ARRAY, convertedtype=UTF8"`
|
||||
SpentOutputIndex int64 `parquet:"name=spent_output_index, type=INT64, repetitiontype=OPTIONAL"`
|
||||
Sequence int64 `parquet:"name=sequence, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// Address string `parquet:"name=address, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// Index int64 `parquet:"name=index, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// RequiredSignatures int64 `parquet:"name=required_signatures, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// ScriptAsm string `parquet:"name=script_asm, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// Type string `parquet:"name=type, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// Value float64 `parquet:"name=value, type=DOUBLE, repetitiontype=OPTIONAL"`
|
||||
}
|
||||
awsTxOutput struct {
|
||||
Script_hex string `parquet:"name=script_hex, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
Value float64 `parquet:"name=value, type=DOUBLE, repetitiontype=OPTIONAL"`
|
||||
// Address string `parquet:"name=address, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// Index int64 `parquet:"name=index, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// Required_signatures int64 `parquet:"name=required_signatures, type=INT64, repetitiontype=OPTIONAL"`
|
||||
// Script_asm string `parquet:"name=script_asm, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
// Type string `parquet:"name=type, type=BYTE_ARRAY, convertedtype=UTF8, repetitiontype=OPTIONAL"`
|
||||
}
|
||||
|
||||
awsFile struct {
|
||||
Key string
|
||||
Size int64
|
||||
LastModified time.Time
|
||||
}
|
||||
)
|
||||
|
||||
func (a awsBlock) ToBlockHeader() (types.BlockHeader, error) {
|
||||
hash, err := chainhash.NewHashFromStr(a.Hash)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Wrap(err, "can't convert block hash")
|
||||
}
|
||||
prevBlockHash, err := chainhash.NewHashFromStr(a.PreviousBlockHash)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Wrap(err, "can't convert previous block hash")
|
||||
}
|
||||
merkleRoot, err := chainhash.NewHashFromStr(a.MerkleRoot)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Wrap(err, "can't convert merkle root")
|
||||
}
|
||||
|
||||
bits, err := strconv.ParseUint(a.Bits, 16, 32)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Wrap(err, "can't convert bits from hex str to uint32")
|
||||
}
|
||||
|
||||
return types.BlockHeader{
|
||||
Hash: *hash,
|
||||
Height: a.Number,
|
||||
Version: int32(a.Version),
|
||||
PrevBlock: *prevBlockHash,
|
||||
MerkleRoot: *merkleRoot,
|
||||
Timestamp: parquettypes.INT96ToTime(a.Timestamp),
|
||||
Bits: uint32(bits),
|
||||
Nonce: uint32(a.Nonce),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a awsTransaction) ToTransaction(block awsBlock) (*types.Transaction, error) {
|
||||
blockhash, err := chainhash.NewHashFromStr(block.Hash)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't convert block hash")
|
||||
}
|
||||
msgtx, err := a.MsgTx(block)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't convert aws tx to wire.msgtx")
|
||||
}
|
||||
return types.ParseMsgTx(msgtx, a.BlockNumber, *blockhash, uint32(a.Index)), nil
|
||||
}
|
||||
|
||||
func (a awsTransaction) MsgTx(block awsBlock) (*wire.MsgTx, error) {
|
||||
txIn := make([]*wire.TxIn, 0, len(a.Inputs))
|
||||
txOut := make([]*wire.TxOut, 0, len(a.Outputs))
|
||||
|
||||
// NOTE: coinbase tx from AWS S3 has no inputs, so we need to add it manually,
|
||||
// but we can't guarantee this data is correct especially the sequence number.
|
||||
if a.IsCoinbase && len(a.Inputs) == 0 {
|
||||
scriptsig, err := hex.DecodeString(block.CoinbaseParam)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't decode script hex")
|
||||
}
|
||||
|
||||
txIn = append(txIn, &wire.TxIn{
|
||||
PreviousOutPoint: wire.OutPoint{
|
||||
Hash: common.ZeroHash,
|
||||
Index: math.MaxUint32,
|
||||
},
|
||||
SignatureScript: scriptsig,
|
||||
Witness: btcutils.CoinbaseWitness,
|
||||
Sequence: math.MaxUint32, // most coinbase tx are using max sequence number
|
||||
})
|
||||
}
|
||||
|
||||
for _, in := range a.Inputs {
|
||||
scriptsig, err := hex.DecodeString(in.ScriptHex)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't decode script hex")
|
||||
}
|
||||
|
||||
witness, err := btcutils.WitnessFromHex(lo.Map(in.TxInWitness, func(src *string, _ int) string {
|
||||
if src == nil {
|
||||
return ""
|
||||
}
|
||||
return *src
|
||||
}))
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't convert witness")
|
||||
}
|
||||
|
||||
prevOutHash, err := chainhash.NewHashFromStr(in.SpentTransactionHash)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't convert prevout hash")
|
||||
}
|
||||
|
||||
txIn = append(txIn, &wire.TxIn{
|
||||
PreviousOutPoint: wire.OutPoint{
|
||||
Hash: *prevOutHash,
|
||||
Index: uint32(in.SpentOutputIndex),
|
||||
},
|
||||
SignatureScript: scriptsig,
|
||||
Witness: witness,
|
||||
Sequence: uint32(in.Sequence),
|
||||
})
|
||||
}
|
||||
|
||||
for _, out := range a.Outputs {
|
||||
scriptpubkey, err := hex.DecodeString(out.Script_hex)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't decode script hex")
|
||||
}
|
||||
txOut = append(txOut, &wire.TxOut{
|
||||
Value: btcutils.BitcoinToSatoshi(out.Value),
|
||||
PkScript: scriptpubkey,
|
||||
})
|
||||
}
|
||||
|
||||
return &wire.MsgTx{
|
||||
Version: int32(a.Version),
|
||||
TxIn: txIn,
|
||||
TxOut: txOut,
|
||||
LockTime: uint32(a.LockTime),
|
||||
}, nil
|
||||
}
|
||||
@@ -24,7 +24,7 @@ const (
|
||||
)
|
||||
|
||||
// Make sure to implement the BitcoinDatasource interface
|
||||
var _ Datasource[[]*types.Block] = (*BitcoinNodeDatasource)(nil)
|
||||
var _ Datasource[*types.Block] = (*BitcoinNodeDatasource)(nil)
|
||||
|
||||
// BitcoinNodeDatasource fetch data from Bitcoin node for Bitcoin Indexer
|
||||
type BitcoinNodeDatasource struct {
|
||||
@@ -88,7 +88,7 @@ func (d *BitcoinNodeDatasource) FetchAsync(ctx context.Context, from, to int64,
|
||||
slogx.String("datasource", d.Name()),
|
||||
)
|
||||
|
||||
from, to, skip, err := d.prepareRange(from, to)
|
||||
from, to, skip, err := d.prepareRange(ctx, from, to)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to prepare fetch range")
|
||||
}
|
||||
@@ -212,12 +212,12 @@ func (d *BitcoinNodeDatasource) FetchAsync(ctx context.Context, from, to int64,
|
||||
return subscription.Client(), nil
|
||||
}
|
||||
|
||||
func (d *BitcoinNodeDatasource) prepareRange(fromHeight, toHeight int64) (start, end int64, skip bool, err error) {
|
||||
func (d *BitcoinNodeDatasource) prepareRange(ctx context.Context, fromHeight, toHeight int64) (start, end int64, skip bool, err error) {
|
||||
start = fromHeight
|
||||
end = toHeight
|
||||
|
||||
// get current bitcoin block height
|
||||
latestBlockHeight, err := d.btcclient.GetBlockCount()
|
||||
latestBlockHeight, err := d.GetCurrentBlockHeight(ctx)
|
||||
if err != nil {
|
||||
return -1, -1, false, errors.Wrap(err, "failed to get block count")
|
||||
}
|
||||
@@ -227,7 +227,7 @@ func (d *BitcoinNodeDatasource) prepareRange(fromHeight, toHeight int64) (start,
|
||||
start = 0
|
||||
}
|
||||
|
||||
// set end to current bitcoin block height if
|
||||
// set end to current bitcoin block height if d
|
||||
// - end is -1
|
||||
// - end is greater that current bitcoin block height
|
||||
if end < 0 || end > latestBlockHeight {
|
||||
@@ -292,3 +292,12 @@ func (d *BitcoinNodeDatasource) GetBlockHeader(ctx context.Context, height int64
|
||||
|
||||
return types.ParseMsgBlockHeader(*block, height), nil
|
||||
}
|
||||
|
||||
// GetCurrentBlockHeight fetch current block height from Bitcoin node
|
||||
func (d *BitcoinNodeDatasource) GetCurrentBlockHeight(ctx context.Context) (int64, error) {
|
||||
height, err := d.btcclient.GetBlockCount()
|
||||
if err != nil {
|
||||
return -1, errors.Wrap(err, "failed to get block height")
|
||||
}
|
||||
return height, nil
|
||||
}
|
||||
|
||||
@@ -10,7 +10,8 @@ import (
|
||||
// Datasource is an interface for indexer data sources.
|
||||
type Datasource[T any] interface {
|
||||
Name() string
|
||||
Fetch(ctx context.Context, from, to int64) (T, error)
|
||||
FetchAsync(ctx context.Context, from, to int64, ch chan<- T) (*subscription.ClientSubscription[T], error)
|
||||
Fetch(ctx context.Context, from, to int64) ([]T, error)
|
||||
FetchAsync(ctx context.Context, from, to int64, ch chan<- []T) (*subscription.ClientSubscription[[]T], error)
|
||||
GetBlockHeader(ctx context.Context, height int64) (types.BlockHeader, error)
|
||||
GetCurrentBlockHeight(ctx context.Context) (int64, error)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package indexers
|
||||
package indexer
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -16,20 +16,15 @@ import (
|
||||
|
||||
const (
|
||||
maxReorgLookBack = 1000
|
||||
|
||||
// pollingInterval is the default polling interval for the indexer polling worker
|
||||
pollingInterval = 15 * time.Second
|
||||
)
|
||||
|
||||
type (
|
||||
BitcoinProcessor Processor[[]*types.Block]
|
||||
BitcoinDatasource datasources.Datasource[[]*types.Block]
|
||||
)
|
||||
|
||||
// Make sure to implement the IndexerWorker interface
|
||||
var _ IndexerWorker = (*BitcoinIndexer)(nil)
|
||||
|
||||
// BitcoinIndexer is the polling indexer for sync Bitcoin data to the database.
|
||||
type BitcoinIndexer struct {
|
||||
Processor BitcoinProcessor
|
||||
Datasource BitcoinDatasource
|
||||
// Indexer generic indexer for fetching and processing data
|
||||
type Indexer[T Input] struct {
|
||||
Processor Processor[T]
|
||||
Datasource datasources.Datasource[T]
|
||||
currentBlock types.BlockHeader
|
||||
|
||||
quitOnce sync.Once
|
||||
@@ -37,9 +32,9 @@ type BitcoinIndexer struct {
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// NewBitcoinIndexer create new BitcoinIndexer
|
||||
func NewBitcoinIndexer(processor BitcoinProcessor, datasource BitcoinDatasource) *BitcoinIndexer {
|
||||
return &BitcoinIndexer{
|
||||
// New create new generic indexer
|
||||
func New[T Input](processor Processor[T], datasource datasources.Datasource[T]) *Indexer[T] {
|
||||
return &Indexer[T]{
|
||||
Processor: processor,
|
||||
Datasource: datasource,
|
||||
|
||||
@@ -48,21 +43,17 @@ func NewBitcoinIndexer(processor BitcoinProcessor, datasource BitcoinDatasource)
|
||||
}
|
||||
}
|
||||
|
||||
func (*BitcoinIndexer) Type() string {
|
||||
return "bitcoin"
|
||||
}
|
||||
|
||||
func (i *BitcoinIndexer) Shutdown() error {
|
||||
func (i *Indexer[T]) Shutdown() error {
|
||||
return i.ShutdownWithContext(context.Background())
|
||||
}
|
||||
|
||||
func (i *BitcoinIndexer) ShutdownWithTimeout(timeout time.Duration) error {
|
||||
func (i *Indexer[T]) ShutdownWithTimeout(timeout time.Duration) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
return i.ShutdownWithContext(ctx)
|
||||
}
|
||||
|
||||
func (i *BitcoinIndexer) ShutdownWithContext(ctx context.Context) (err error) {
|
||||
func (i *Indexer[T]) ShutdownWithContext(ctx context.Context) (err error) {
|
||||
i.quitOnce.Do(func() {
|
||||
close(i.quit)
|
||||
select {
|
||||
@@ -76,12 +67,11 @@ func (i *BitcoinIndexer) ShutdownWithContext(ctx context.Context) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (i *BitcoinIndexer) Run(ctx context.Context) (err error) {
|
||||
func (i *Indexer[T]) Run(ctx context.Context) (err error) {
|
||||
defer close(i.done)
|
||||
|
||||
ctx = logger.WithContext(ctx,
|
||||
slog.String("package", "indexers"),
|
||||
slog.String("indexer", i.Type()),
|
||||
slog.String("processor", i.Processor.Name()),
|
||||
slog.String("datasource", i.Datasource.Name()),
|
||||
)
|
||||
@@ -114,15 +104,15 @@ func (i *BitcoinIndexer) Run(ctx context.Context) (err error) {
|
||||
}
|
||||
}
|
||||
|
||||
func (i *BitcoinIndexer) process(ctx context.Context) (err error) {
|
||||
func (i *Indexer[T]) process(ctx context.Context) (err error) {
|
||||
// height range to fetch data
|
||||
from, to := i.currentBlock.Height+1, int64(-1)
|
||||
|
||||
logger.InfoContext(ctx, "Start fetching bitcoin blocks", slog.Int64("from", from))
|
||||
ch := make(chan []*types.Block)
|
||||
logger.InfoContext(ctx, "Start fetching input data", slog.Int64("from", from))
|
||||
ch := make(chan []T)
|
||||
subscription, err := i.Datasource.FetchAsync(ctx, from, to, ch)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to fetch data")
|
||||
return errors.Wrap(err, "failed to fetch input data")
|
||||
}
|
||||
defer subscription.Unsubscribe()
|
||||
|
||||
@@ -130,21 +120,24 @@ func (i *BitcoinIndexer) process(ctx context.Context) (err error) {
|
||||
select {
|
||||
case <-i.quit:
|
||||
return nil
|
||||
case blocks := <-ch:
|
||||
// empty blocks
|
||||
if len(blocks) == 0 {
|
||||
case inputs := <-ch:
|
||||
// empty inputs
|
||||
if len(inputs) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
firstInput := inputs[0]
|
||||
firstInputHeader := firstInput.BlockHeader()
|
||||
|
||||
startAt := time.Now()
|
||||
ctx := logger.WithContext(ctx,
|
||||
slogx.Int64("from", blocks[0].Header.Height),
|
||||
slogx.Int64("to", blocks[len(blocks)-1].Header.Height),
|
||||
slogx.Int64("from", firstInputHeader.Height),
|
||||
slogx.Int64("to", inputs[len(inputs)-1].BlockHeader().Height),
|
||||
)
|
||||
|
||||
// validate reorg from first block
|
||||
// validate reorg from first input
|
||||
{
|
||||
remoteBlockHeader := blocks[0].Header
|
||||
remoteBlockHeader := firstInputHeader
|
||||
if !remoteBlockHeader.PrevBlock.IsEqual(&i.currentBlock.Hash) {
|
||||
logger.WarnContext(ctx, "Detected chain reorganization. Searching for fork point...",
|
||||
slogx.String("event", "reorg_detected"),
|
||||
@@ -210,33 +203,35 @@ func (i *BitcoinIndexer) process(ctx context.Context) (err error) {
|
||||
}
|
||||
}
|
||||
|
||||
// validate is block is continuous and no reorg
|
||||
for i := 1; i < len(blocks); i++ {
|
||||
if blocks[i].Header.Height != blocks[i-1].Header.Height+1 {
|
||||
return errors.Wrapf(errs.InternalError, "block is not continuous, block[%d] height: %d, block[%d] height: %d", i-1, blocks[i-1].Header.Height, i, blocks[i].Header.Height)
|
||||
// validate is input is continuous and no reorg
|
||||
for i := 1; i < len(inputs); i++ {
|
||||
header := inputs[i].BlockHeader()
|
||||
prevHeader := inputs[i-1].BlockHeader()
|
||||
if header.Height != prevHeader.Height+1 {
|
||||
return errors.Wrapf(errs.InternalError, "input is not continuous, input[%d] height: %d, input[%d] height: %d", i-1, prevHeader.Height, i, header.Height)
|
||||
}
|
||||
|
||||
if !blocks[i].Header.PrevBlock.IsEqual(&blocks[i-1].Header.Hash) {
|
||||
logger.WarnContext(ctx, "Chain Reorganization occurred in the middle of batch fetching blocks, need to try to fetch again")
|
||||
if !header.PrevBlock.IsEqual(&prevHeader.Hash) {
|
||||
logger.WarnContext(ctx, "Chain Reorganization occurred in the middle of batch fetching inputs, need to try to fetch again")
|
||||
|
||||
// end current round
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
ctx = logger.WithContext(ctx, slog.Int("total_blocks", len(blocks)))
|
||||
ctx = logger.WithContext(ctx, slog.Int("total_inputs", len(inputs)))
|
||||
|
||||
// Start processing blocks
|
||||
logger.InfoContext(ctx, "Processing blocks")
|
||||
if err := i.Processor.Process(ctx, blocks); err != nil {
|
||||
// Start processing input
|
||||
logger.InfoContext(ctx, "Processing inputs")
|
||||
if err := i.Processor.Process(ctx, inputs); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
|
||||
// Update current state
|
||||
i.currentBlock = blocks[len(blocks)-1].Header
|
||||
i.currentBlock = inputs[len(inputs)-1].BlockHeader()
|
||||
|
||||
logger.InfoContext(ctx, "Processed blocks successfully",
|
||||
slogx.String("event", "processed_blocks"),
|
||||
logger.InfoContext(ctx, "Processed inputs successfully",
|
||||
slogx.String("event", "processed_inputs"),
|
||||
slogx.Int64("current_block", i.currentBlock.Height),
|
||||
slogx.Duration("duration", time.Since(startAt)),
|
||||
)
|
||||
@@ -1,4 +1,4 @@
|
||||
package indexers
|
||||
package indexer
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -7,24 +7,15 @@ import (
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
)
|
||||
|
||||
const (
|
||||
// pollingInterval is the default polling interval for the indexer polling worker
|
||||
pollingInterval = 15 * time.Second
|
||||
)
|
||||
|
||||
type IndexerWorker interface {
|
||||
Type() string
|
||||
Run(ctx context.Context) error
|
||||
Shutdown() error
|
||||
ShutdownWithTimeout(timeout time.Duration) error
|
||||
ShutdownWithContext(ctx context.Context) error
|
||||
type Input interface {
|
||||
BlockHeader() types.BlockHeader
|
||||
}
|
||||
|
||||
type Processor[T any] interface {
|
||||
type Processor[T Input] interface {
|
||||
Name() string
|
||||
|
||||
// Process processes the input data and indexes it.
|
||||
Process(ctx context.Context, inputs T) error
|
||||
Process(ctx context.Context, inputs []T) error
|
||||
|
||||
// CurrentBlock returns the latest indexed block header.
|
||||
CurrentBlock(ctx context.Context) (types.BlockHeader, error)
|
||||
@@ -39,3 +30,10 @@ type Processor[T any] interface {
|
||||
// to ensure the last shutdown was graceful and no missing data.
|
||||
VerifyStates(ctx context.Context) error
|
||||
}
|
||||
|
||||
type IndexerWorker interface {
|
||||
Shutdown() error
|
||||
ShutdownWithTimeout(timeout time.Duration) error
|
||||
ShutdownWithContext(ctx context.Context) (err error)
|
||||
Run(ctx context.Context) (err error)
|
||||
}
|
||||
@@ -38,6 +38,10 @@ type Block struct {
|
||||
Transactions []*Transaction
|
||||
}
|
||||
|
||||
func (b *Block) BlockHeader() BlockHeader {
|
||||
return b.Header
|
||||
}
|
||||
|
||||
func ParseMsgBlock(src *wire.MsgBlock, height int64) *Block {
|
||||
hash := src.Header.BlockHash()
|
||||
return &Block{
|
||||
|
||||
1
docker/.gitignore
vendored
1
docker/.gitignore
vendored
@@ -1 +0,0 @@
|
||||
volumes
|
||||
@@ -1,43 +0,0 @@
|
||||
logger:
|
||||
output: text
|
||||
debug: false
|
||||
|
||||
bitcoin_node:
|
||||
host: "bitcoin-mainnet-archive.allthatnode.com"
|
||||
user: "user"
|
||||
pass: "pass"
|
||||
disable_tls: false
|
||||
|
||||
network: mainnet
|
||||
|
||||
reporting:
|
||||
disabled: false
|
||||
base_url: "https://staging.indexer.api.gaze.network" # defaults to "https://indexer.api.gaze.network" if empty
|
||||
name: "Local Indexer"
|
||||
website_url: "" # public website URL to show on the dashboard. Can be left empty.
|
||||
indexer_api_url: "" # public url to access this api. Can be left empty.
|
||||
|
||||
้http_server:
|
||||
port: 8080
|
||||
|
||||
modules:
|
||||
bitcoin:
|
||||
database: "postgres" # Store bitcoin data in postgres
|
||||
postgres:
|
||||
host: "db"
|
||||
port: 5432
|
||||
user: "postgres"
|
||||
password: "password"
|
||||
db_name: "postgres"
|
||||
|
||||
runes:
|
||||
database: "postgres" # Store Runes data in postgres
|
||||
datasource: "postgres" # Fetch bitcoin blocks from postgres
|
||||
api_handlers:
|
||||
- http
|
||||
postgres:
|
||||
host: "db"
|
||||
port: 5432
|
||||
user: "postgres"
|
||||
password: "password"
|
||||
db_name: "postgres"
|
||||
@@ -1,47 +0,0 @@
|
||||
services:
|
||||
# TODO: need to mount migrations folder
|
||||
gaze-migrator:
|
||||
# image: ghcr.io/gaze-network/gaze-indexer:v0.1.0
|
||||
build:
|
||||
context: ../
|
||||
dockerfile: ./docker/Dockerfile
|
||||
restart: on-failure
|
||||
depends_on:
|
||||
- db
|
||||
networks:
|
||||
- indexer
|
||||
command:
|
||||
["/app/main", "migrate", "up", "--bitcoin", "--runes", "--database", "postgres://postgres:password@db:5432/postgres?sslmode=disable"]
|
||||
|
||||
gaze-indexer:
|
||||
# image: ghcr.io/gaze-network/gaze-indexer:v0.1.0
|
||||
build:
|
||||
context: ../
|
||||
dockerfile: ./docker/Dockerfile
|
||||
restart: unless-stopped
|
||||
depends_on:
|
||||
- db
|
||||
- gaze-migrator:
|
||||
condition: service_completed_successfully
|
||||
networks:
|
||||
- indexer
|
||||
ports:
|
||||
- 8080:8080
|
||||
volumes:
|
||||
- "./config.example.yaml:/app/config.yaml"
|
||||
command: ["/app/main", "run", "--bitcoin", "--runes"]
|
||||
|
||||
db:
|
||||
image: postgres:16-alpine
|
||||
volumes:
|
||||
- "./volumes/postgresql/data:/var/lib/postgresql/data"
|
||||
networks:
|
||||
- indexer
|
||||
environment:
|
||||
- POSTGRES_USER=postgres
|
||||
- POSTGRES_PASSWORD=password
|
||||
- POSTGRES_DB=postgres
|
||||
|
||||
networks:
|
||||
indexer:
|
||||
driver: bridge
|
||||
33
go.mod
33
go.mod
@@ -4,6 +4,10 @@ go 1.22
|
||||
|
||||
require (
|
||||
github.com/Cleverse/go-utilities/utils v0.0.0-20240119201306-d71eb577ef11
|
||||
github.com/aws/aws-sdk-go-v2 v1.23.0
|
||||
github.com/aws/aws-sdk-go-v2/config v1.25.3
|
||||
github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.14.0
|
||||
github.com/aws/aws-sdk-go-v2/service/s3 v1.43.0
|
||||
github.com/btcsuite/btcd v0.24.0
|
||||
github.com/btcsuite/btcd/btcutil v1.1.5
|
||||
github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0
|
||||
@@ -11,23 +15,42 @@ require (
|
||||
github.com/gaze-network/uint128 v1.3.0
|
||||
github.com/gofiber/fiber/v2 v2.52.4
|
||||
github.com/golang-migrate/migrate/v4 v4.17.1
|
||||
github.com/jackc/pgx v3.6.2+incompatible
|
||||
github.com/jackc/pgx/v5 v5.5.5
|
||||
github.com/mcosta74/pgx-slog v0.3.0
|
||||
github.com/planxnx/concurrent-stream v0.1.5
|
||||
github.com/samber/do/v2 v2.0.0-beta.7
|
||||
github.com/samber/lo v1.39.0
|
||||
github.com/shopspring/decimal v1.3.1
|
||||
github.com/shopspring/decimal v1.4.0
|
||||
github.com/spf13/cobra v1.8.0
|
||||
github.com/spf13/pflag v1.0.5
|
||||
github.com/spf13/viper v1.18.2
|
||||
github.com/stretchr/testify v1.8.4
|
||||
github.com/valyala/fasthttp v1.51.0
|
||||
github.com/xitongsys/parquet-go v1.6.2
|
||||
github.com/xitongsys/parquet-go-source v0.0.0-20240122235623-d6294584ab18
|
||||
go.uber.org/automaxprocs v1.5.3
|
||||
golang.org/x/sync v0.5.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/andybalholm/brotli v1.0.5 // indirect
|
||||
github.com/apache/arrow/go/arrow v0.0.0-20200730104253-651201b0f516 // indirect
|
||||
github.com/apache/thrift v0.20.0 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.16.2 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.17.2 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.0 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.25.3 // indirect
|
||||
github.com/aws/smithy-go v1.17.0 // indirect
|
||||
github.com/btcsuite/btcd/btcec/v2 v2.1.3 // indirect
|
||||
github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f // indirect
|
||||
github.com/btcsuite/go-socks v0.0.0-20170105172521-4720035b7bfd // indirect
|
||||
@@ -40,6 +63,7 @@ require (
|
||||
github.com/fsnotify/fsnotify v1.7.0 // indirect
|
||||
github.com/getsentry/sentry-go v0.18.0 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/golang/snappy v0.0.4 // indirect
|
||||
github.com/google/uuid v1.5.0 // indirect
|
||||
github.com/hashicorp/errwrap v1.1.0 // indirect
|
||||
github.com/hashicorp/go-multierror v1.1.1 // indirect
|
||||
@@ -47,7 +71,9 @@ require (
|
||||
github.com/inconshreveable/mousetrap v1.1.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
|
||||
github.com/jackc/pgx v3.6.2+incompatible // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.1 // indirect
|
||||
github.com/jmespath/go-jmespath v0.4.0 // indirect
|
||||
github.com/klauspost/compress v1.17.0 // indirect
|
||||
github.com/kr/pretty v0.3.1 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
@@ -58,12 +84,14 @@ require (
|
||||
github.com/mattn/go-runewidth v0.0.15 // indirect
|
||||
github.com/mitchellh/mapstructure v1.5.0 // indirect
|
||||
github.com/pelletier/go-toml/v2 v2.1.0 // indirect
|
||||
github.com/pierrec/lz4/v4 v4.1.16 // indirect
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
|
||||
github.com/rivo/uniseg v0.2.0 // indirect
|
||||
github.com/rogpeppe/go-internal v1.9.0 // indirect
|
||||
github.com/sagikazarmark/locafero v0.4.0 // indirect
|
||||
github.com/sagikazarmark/slog-shim v0.1.0 // indirect
|
||||
github.com/samber/go-type-to-string v1.4.0 // indirect
|
||||
github.com/sourcegraph/conc v0.3.0 // indirect
|
||||
github.com/spf13/afero v1.11.0 // indirect
|
||||
github.com/spf13/cast v1.6.0 // indirect
|
||||
@@ -76,6 +104,7 @@ require (
|
||||
golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
|
||||
golang.org/x/sys v0.17.0 // indirect
|
||||
golang.org/x/text v0.14.0 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
|
||||
gopkg.in/ini.v1 v1.67.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
)
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
btcconfig "github.com/gaze-network/indexer-network/modules/bitcoin/config"
|
||||
runesconfig "github.com/gaze-network/indexer-network/modules/runes/config"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger/slogx"
|
||||
@@ -25,20 +24,31 @@ var (
|
||||
Output: "TEXT",
|
||||
},
|
||||
Network: common.NetworkMainnet,
|
||||
HTTPServer: HTTPServerConfig{
|
||||
Port: 8080,
|
||||
},
|
||||
BitcoinNode: BitcoinNodeClient{
|
||||
User: "user",
|
||||
Pass: "pass",
|
||||
},
|
||||
Modules: Modules{
|
||||
Runes: runesconfig.Config{
|
||||
Datasource: "bitcoin-node",
|
||||
Database: "postgres",
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Logger logger.Config `mapstructure:"logger"`
|
||||
BitcoinNode BitcoinNodeClient `mapstructure:"bitcoin_node"`
|
||||
Network common.Network `mapstructure:"network"`
|
||||
HTTPServer HTTPServerConfig `mapstructure:"http_server"`
|
||||
Modules Modules `mapstructure:"modules"`
|
||||
Reporting reportingclient.Config `mapstructure:"reporting"`
|
||||
EnableModules []string `mapstructure:"enable_modules"`
|
||||
APIOnly bool `mapstructure:"api_only"`
|
||||
Logger logger.Config `mapstructure:"logger"`
|
||||
BitcoinNode BitcoinNodeClient `mapstructure:"bitcoin_node"`
|
||||
Network common.Network `mapstructure:"network"`
|
||||
HTTPServer HTTPServerConfig `mapstructure:"http_server"`
|
||||
Modules Modules `mapstructure:"modules"`
|
||||
Reporting reportingclient.Config `mapstructure:"reporting"`
|
||||
}
|
||||
|
||||
type BitcoinNodeClient struct {
|
||||
@@ -49,8 +59,7 @@ type BitcoinNodeClient struct {
|
||||
}
|
||||
|
||||
type Modules struct {
|
||||
Bitcoin btcconfig.Config `mapstructure:"bitcoin"`
|
||||
Runes runesconfig.Config `mapstructure:"runes"`
|
||||
Runes runesconfig.Config `mapstructure:"runes"`
|
||||
}
|
||||
|
||||
type HTTPServerConfig struct {
|
||||
|
||||
@@ -40,7 +40,7 @@ func New(ctx context.Context, conf Config) (*pgx.Conn, error) {
|
||||
// Prepare connection pool configuration
|
||||
connConfig, err := pgx.ParseConfig(conf.String())
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to parse config to create a new connection")
|
||||
return nil, errors.Join(errs.InvalidArgument, errors.Wrap(err, "failed while parse config"))
|
||||
}
|
||||
connConfig.Tracer = conf.QueryTracer()
|
||||
|
||||
|
||||
1
main.go
1
main.go
@@ -7,7 +7,6 @@ import (
|
||||
"syscall"
|
||||
|
||||
"github.com/gaze-network/indexer-network/cmd"
|
||||
_ "go.uber.org/automaxprocs"
|
||||
)
|
||||
|
||||
func main() {
|
||||
|
||||
@@ -1,244 +0,0 @@
|
||||
package btcclient
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/datasources"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/internal/subscription"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/datagateway"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger/slogx"
|
||||
cstream "github.com/planxnx/concurrent-stream"
|
||||
"github.com/samber/lo"
|
||||
)
|
||||
|
||||
// TODO: Refactor this, datasources.BitcoinNode and This package is the same.
|
||||
|
||||
const (
|
||||
blockStreamChunkSize = 100
|
||||
)
|
||||
|
||||
// Make sure to implement the BitcoinDatasource interface
|
||||
var _ datasources.Datasource[[]*types.Block] = (*ClientDatabase)(nil)
|
||||
|
||||
// ClientDatabase is a client to connect to the bitcoin database.
|
||||
type ClientDatabase struct {
|
||||
bitcoinDg datagateway.BitcoinDataGateway
|
||||
}
|
||||
|
||||
func NewClientDatabase(bitcoinDg datagateway.BitcoinDataGateway) *ClientDatabase {
|
||||
return &ClientDatabase{
|
||||
bitcoinDg: bitcoinDg,
|
||||
}
|
||||
}
|
||||
|
||||
func (d ClientDatabase) Name() string {
|
||||
return "bitcoin_database"
|
||||
}
|
||||
|
||||
func (d *ClientDatabase) Fetch(ctx context.Context, from, to int64) ([]*types.Block, error) {
|
||||
ch := make(chan []*types.Block)
|
||||
subscription, err := d.FetchAsync(ctx, from, to, ch)
|
||||
if err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
}
|
||||
defer subscription.Unsubscribe()
|
||||
|
||||
blocks := make([]*types.Block, 0)
|
||||
for {
|
||||
select {
|
||||
case b, ok := <-ch:
|
||||
if !ok {
|
||||
return blocks, nil
|
||||
}
|
||||
blocks = append(blocks, b...)
|
||||
case <-subscription.Done():
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, errors.Wrap(err, "context done")
|
||||
}
|
||||
return blocks, nil
|
||||
case err := <-subscription.Err():
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "got error while fetch async")
|
||||
}
|
||||
return blocks, nil
|
||||
case <-ctx.Done():
|
||||
return nil, errors.Wrap(ctx.Err(), "context done")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *ClientDatabase) FetchAsync(ctx context.Context, from, to int64, ch chan<- []*types.Block) (*subscription.ClientSubscription[[]*types.Block], error) {
|
||||
ctx = logger.WithContext(ctx,
|
||||
slogx.String("package", "datasources"),
|
||||
slogx.String("datasource", d.Name()),
|
||||
)
|
||||
|
||||
from, to, skip, err := d.prepareRange(ctx, from, to)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to prepare fetch range")
|
||||
}
|
||||
|
||||
subscription := subscription.NewSubscription(ch)
|
||||
if skip {
|
||||
if err := subscription.UnsubscribeWithContext(ctx); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to unsubscribe")
|
||||
}
|
||||
return subscription.Client(), nil
|
||||
}
|
||||
|
||||
// Create parallel stream
|
||||
out := make(chan []*types.Block)
|
||||
stream := cstream.NewStream(ctx, 8, out)
|
||||
|
||||
// create slice of block height to fetch
|
||||
blockHeights := make([]int64, 0, to-from+1)
|
||||
for i := from; i <= to; i++ {
|
||||
blockHeights = append(blockHeights, i)
|
||||
}
|
||||
|
||||
// Wait for stream to finish and close out channel
|
||||
go func() {
|
||||
defer close(out)
|
||||
_ = stream.Wait()
|
||||
}()
|
||||
|
||||
// Fan-out blocks to subscription channel
|
||||
go func() {
|
||||
defer func() {
|
||||
// add a bit delay to prevent shutdown before client receive all blocks
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
subscription.Unsubscribe()
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
case data, ok := <-out:
|
||||
// stream closed
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
// empty blocks
|
||||
if len(data) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// send blocks to subscription channel
|
||||
if err := subscription.Send(ctx, data); err != nil {
|
||||
if errors.Is(err, errs.Closed) {
|
||||
return
|
||||
}
|
||||
logger.WarnContext(ctx, "Failed to send bitcoin blocks to subscription client",
|
||||
slogx.Int64("start", data[0].Header.Height),
|
||||
slogx.Int64("end", data[len(data)-1].Header.Height),
|
||||
slogx.Error(err),
|
||||
)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Parallel fetch blocks from Bitcoin node until complete all block heights
|
||||
// or subscription is done.
|
||||
go func() {
|
||||
defer stream.Close()
|
||||
done := subscription.Done()
|
||||
chunks := lo.Chunk(blockHeights, blockStreamChunkSize)
|
||||
for _, chunk := range chunks {
|
||||
chunk := chunk
|
||||
select {
|
||||
case <-done:
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
if len(chunk) == 0 {
|
||||
continue
|
||||
}
|
||||
stream.Go(func() []*types.Block {
|
||||
startAt := time.Now()
|
||||
defer func() {
|
||||
logger.DebugContext(ctx, "Fetched chunk of blocks from Bitcoin node",
|
||||
slogx.Int("total_blocks", len(chunk)),
|
||||
slogx.Int64("from", chunk[0]),
|
||||
slogx.Int64("to", chunk[len(chunk)-1]),
|
||||
slogx.Duration("duration", time.Since(startAt)),
|
||||
)
|
||||
}()
|
||||
|
||||
fromHeight, toHeight := chunk[0], chunk[len(chunk)-1]
|
||||
blocks, err := d.bitcoinDg.GetBlocksByHeightRange(ctx, fromHeight, toHeight)
|
||||
if err != nil {
|
||||
logger.ErrorContext(ctx, "Can't get block data from Bitcoin database",
|
||||
slogx.Error(err),
|
||||
slogx.Int64("from", fromHeight),
|
||||
slogx.Int64("to", toHeight),
|
||||
)
|
||||
if err := subscription.SendError(ctx, errors.Wrapf(err, "failed to get blocks: from_height: %d, to_height: %d", fromHeight, toHeight)); err != nil {
|
||||
logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return blocks
|
||||
})
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return subscription.Client(), nil
|
||||
}
|
||||
|
||||
func (c *ClientDatabase) GetBlockHeader(ctx context.Context, height int64) (types.BlockHeader, error) {
|
||||
header, err := c.bitcoinDg.GetBlockHeaderByHeight(ctx, height)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.WithStack(err)
|
||||
}
|
||||
return header, nil
|
||||
}
|
||||
|
||||
func (c *ClientDatabase) prepareRange(ctx context.Context, fromHeight, toHeight int64) (start, end int64, skip bool, err error) {
|
||||
start = fromHeight
|
||||
end = toHeight
|
||||
|
||||
// get current bitcoin block height
|
||||
latestBlock, err := c.bitcoinDg.GetLatestBlockHeader(ctx)
|
||||
if err != nil {
|
||||
return -1, -1, false, errors.Wrap(err, "failed to get block count")
|
||||
}
|
||||
|
||||
// set start to genesis block height
|
||||
if start < 0 {
|
||||
start = 0
|
||||
}
|
||||
|
||||
// set end to current bitcoin block height if
|
||||
// - end is -1
|
||||
// - end is greater that current bitcoin block height
|
||||
if end < 0 || end > latestBlock.Height {
|
||||
end = latestBlock.Height
|
||||
}
|
||||
|
||||
// if start is greater than end, skip this round
|
||||
if start > end {
|
||||
return -1, -1, true, nil
|
||||
}
|
||||
|
||||
return start, end, false, nil
|
||||
}
|
||||
|
||||
// GetTransactionByHash returns a transaction with the given hash. Returns errs.NotFound if transaction does not exist.
|
||||
func (c *ClientDatabase) GetTransactionByHash(ctx context.Context, txHash chainhash.Hash) (*types.Transaction, error) {
|
||||
tx, err := c.bitcoinDg.GetTransactionByHash(ctx, txHash)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to get transaction by hash")
|
||||
}
|
||||
return tx, nil
|
||||
}
|
||||
@@ -1,8 +0,0 @@
|
||||
package config
|
||||
|
||||
import "github.com/gaze-network/indexer-network/internal/postgres"
|
||||
|
||||
type Config struct {
|
||||
Database string `mapstructure:"database"` // Database to store bitcoin data.
|
||||
Postgres postgres.Config `mapstructure:"postgres"`
|
||||
}
|
||||
@@ -1,26 +0,0 @@
|
||||
package bitcoin
|
||||
|
||||
import (
|
||||
"github.com/Cleverse/go-utilities/utils"
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
)
|
||||
|
||||
const (
|
||||
Version = "v0.0.1"
|
||||
DBVersion = 1
|
||||
)
|
||||
|
||||
var (
|
||||
// defaultCurrentBlockHeight is the default value for the current block height for first time indexing
|
||||
defaultCurrentBlock = types.BlockHeader{
|
||||
Hash: common.ZeroHash,
|
||||
Height: -1,
|
||||
}
|
||||
|
||||
lastV1Block = types.BlockHeader{
|
||||
Hash: *utils.Must(chainhash.NewHashFromStr("00000000000001aa077d7aa84c532a4d69bdbff519609d1da0835261b7a74eb6")),
|
||||
Height: 227835,
|
||||
}
|
||||
)
|
||||
@@ -1,18 +0,0 @@
|
||||
BEGIN;
|
||||
|
||||
-- DROP INDEX
|
||||
DROP INDEX IF EXISTS bitcoin_blocks_block_hash_idx;
|
||||
DROP INDEX IF EXISTS bitcoin_transactions_tx_hash_idx;
|
||||
DROP INDEX IF EXISTS bitcoin_transactions_block_hash_idx;
|
||||
DROP INDEX IF EXISTS bitcoin_transaction_txouts_pkscript_idx;
|
||||
DROP INDEX IF EXISTS bitcoin_transaction_txins_prevout_idx;
|
||||
|
||||
-- DROP TABLE
|
||||
DROP TABLE IF EXISTS "bitcoin_indexer_stats";
|
||||
DROP TABLE IF EXISTS "bitcoin_indexer_db_version";
|
||||
DROP TABLE IF EXISTS "bitcoin_transaction_txins";
|
||||
DROP TABLE IF EXISTS "bitcoin_transaction_txouts";
|
||||
DROP TABLE IF EXISTS "bitcoin_transactions";
|
||||
DROP TABLE IF EXISTS "bitcoin_blocks";
|
||||
|
||||
COMMIT;
|
||||
@@ -1,72 +0,0 @@
|
||||
BEGIN;
|
||||
|
||||
-- Indexer Client Information
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "bitcoin_indexer_stats" (
|
||||
"id" BIGSERIAL PRIMARY KEY,
|
||||
"client_version" TEXT NOT NULL,
|
||||
"network" TEXT NOT NULL,
|
||||
"created_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "bitcoin_indexer_db_version" (
|
||||
"id" BIGSERIAL PRIMARY KEY,
|
||||
"version" INT NOT NULL,
|
||||
"created_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
INSERT INTO "bitcoin_indexer_db_version" ("version") VALUES (1);
|
||||
|
||||
-- Bitcoin Data
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "bitcoin_blocks" (
|
||||
"block_height" INT NOT NULL PRIMARY KEY,
|
||||
"block_hash" TEXT NOT NULL,
|
||||
"version" INT NOT NULL,
|
||||
"merkle_root" TEXT NOT NULL,
|
||||
"prev_block_hash" TEXT NOT NULL,
|
||||
"timestamp" TIMESTAMP WITH TIME ZONE NOT NULL,
|
||||
"bits" BIGINT NOT NULL,
|
||||
"nonce" BIGINT NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bitcoin_blocks_block_hash_idx ON "bitcoin_blocks" USING HASH ("block_hash");
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "bitcoin_transactions" (
|
||||
"tx_hash" TEXT NOT NULL, -- can't use as primary key because block v1 has duplicate tx hashes (coinbase tx). See: https://github.com/bitcoin/bitcoin/commit/a206b0ea12eb4606b93323268fc81a4f1f952531
|
||||
"version" INT NOT NULL,
|
||||
"locktime" BIGINT NOT NULL,
|
||||
"block_height" INT NOT NULL,
|
||||
"block_hash" TEXT NOT NULL,
|
||||
"idx" INT NOT NULL,
|
||||
PRIMARY KEY ("block_height", "idx")
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bitcoin_transactions_tx_hash_idx ON "bitcoin_transactions" USING HASH ("tx_hash");
|
||||
CREATE INDEX IF NOT EXISTS bitcoin_transactions_block_hash_idx ON "bitcoin_transactions" USING HASH ("block_hash");
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "bitcoin_transaction_txouts" (
|
||||
"tx_hash" TEXT NOT NULL,
|
||||
"tx_idx" INT NOT NULL,
|
||||
"pkscript" TEXT NOT NULL, -- Hex String
|
||||
"value" BIGINT NOT NULL,
|
||||
"is_spent" BOOLEAN NOT NULL DEFAULT false,
|
||||
PRIMARY KEY ("tx_hash", "tx_idx")
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bitcoin_transaction_txouts_pkscript_idx ON "bitcoin_transaction_txouts" USING HASH ("pkscript");
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "bitcoin_transaction_txins" (
|
||||
"tx_hash" TEXT NOT NULL,
|
||||
"tx_idx" INT NOT NULL,
|
||||
"prevout_tx_hash" TEXT NOT NULL,
|
||||
"prevout_tx_idx" INT NOT NULL,
|
||||
"prevout_pkscript" TEXT NULL, -- Hex String, Can be NULL if the prevout is a coinbase transaction
|
||||
"scriptsig" TEXT NOT NULL, -- Hex String
|
||||
"witness" TEXT NOT NULL DEFAULT '', -- Hex String
|
||||
"sequence" BIGINT NOT NULL,
|
||||
PRIMARY KEY ("tx_hash", "tx_idx")
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bitcoin_transaction_txins_prevout_idx ON "bitcoin_transaction_txins" USING BTREE ("prevout_tx_hash", "prevout_tx_idx");
|
||||
|
||||
COMMIT;
|
||||
@@ -1,99 +0,0 @@
|
||||
-- name: GetLatestBlockHeader :one
|
||||
SELECT * FROM bitcoin_blocks ORDER BY block_height DESC LIMIT 1;
|
||||
|
||||
-- name: InsertBlock :exec
|
||||
INSERT INTO bitcoin_blocks ("block_height","block_hash","version","merkle_root","prev_block_hash","timestamp","bits","nonce") VALUES ($1, $2, $3, $4, $5, $6, $7, $8);
|
||||
|
||||
-- name: BatchInsertBlocks :exec
|
||||
INSERT INTO bitcoin_blocks ("block_height","block_hash","version","merkle_root","prev_block_hash","timestamp","bits","nonce")
|
||||
VALUES (
|
||||
unnest(@block_height_arr::INT[]),
|
||||
unnest(@block_hash_arr::TEXT[]),
|
||||
unnest(@version_arr::INT[]),
|
||||
unnest(@merkle_root_arr::TEXT[]),
|
||||
unnest(@prev_block_hash_arr::TEXT[]),
|
||||
unnest(@timestamp_arr::TIMESTAMP WITH TIME ZONE[]), -- or use TIMESTAMPTZ
|
||||
unnest(@bits_arr::BIGINT[]),
|
||||
unnest(@nonce_arr::BIGINT[])
|
||||
);
|
||||
|
||||
-- name: BatchInsertTransactions :exec
|
||||
INSERT INTO bitcoin_transactions ("tx_hash","version","locktime","block_height","block_hash","idx")
|
||||
VALUES (
|
||||
unnest(@tx_hash_arr::TEXT[]),
|
||||
unnest(@version_arr::INT[]),
|
||||
unnest(@locktime_arr::BIGINT[]),
|
||||
unnest(@block_height_arr::INT[]),
|
||||
unnest(@block_hash_arr::TEXT[]),
|
||||
unnest(@idx_arr::INT[])
|
||||
);
|
||||
|
||||
-- name: BatchInsertTransactionTxIns :exec
|
||||
WITH update_txout AS (
|
||||
UPDATE "bitcoin_transaction_txouts"
|
||||
SET "is_spent" = true
|
||||
FROM (SELECT unnest(@prevout_tx_hash_arr::TEXT[]) as tx_hash, unnest(@prevout_tx_idx_arr::INT[]) as tx_idx) as txin
|
||||
WHERE "bitcoin_transaction_txouts"."tx_hash" = txin.tx_hash AND "bitcoin_transaction_txouts"."tx_idx" = txin.tx_idx AND "is_spent" = false
|
||||
RETURNING "bitcoin_transaction_txouts"."tx_hash", "bitcoin_transaction_txouts"."tx_idx", "pkscript"
|
||||
), prepare_insert AS (
|
||||
SELECT input.tx_hash, input.tx_idx, prevout_tx_hash, prevout_tx_idx, update_txout.pkscript as prevout_pkscript, scriptsig, witness, sequence
|
||||
FROM (
|
||||
SELECT
|
||||
unnest(@tx_hash_arr::TEXT[]) as tx_hash,
|
||||
unnest(@tx_idx_arr::INT[]) as tx_idx,
|
||||
unnest(@prevout_tx_hash_arr::TEXT[]) as prevout_tx_hash,
|
||||
unnest(@prevout_tx_idx_arr::INT[]) as prevout_tx_idx,
|
||||
unnest(@scriptsig_arr::TEXT[]) as scriptsig,
|
||||
unnest(@witness_arr::TEXT[]) as witness,
|
||||
unnest(@sequence_arr::INT[]) as sequence
|
||||
) input LEFT JOIN update_txout ON "update_txout"."tx_hash" = "input"."prevout_tx_hash" AND "update_txout"."tx_idx" = "input"."prevout_tx_idx"
|
||||
)
|
||||
INSERT INTO bitcoin_transaction_txins ("tx_hash","tx_idx","prevout_tx_hash","prevout_tx_idx", "prevout_pkscript","scriptsig","witness","sequence")
|
||||
SELECT "tx_hash", "tx_idx", "prevout_tx_hash", "prevout_tx_idx", "prevout_pkscript", "scriptsig", "witness", "sequence" FROM prepare_insert;
|
||||
|
||||
-- name: BatchInsertTransactionTxOuts :exec
|
||||
INSERT INTO bitcoin_transaction_txouts ("tx_hash","tx_idx","pkscript","value")
|
||||
VALUES (
|
||||
unnest(@tx_hash_arr::TEXT[]),
|
||||
unnest(@tx_idx_arr::INT[]),
|
||||
unnest(@pkscript_arr::TEXT[]),
|
||||
unnest(@value_arr::BIGINT[])
|
||||
);
|
||||
|
||||
-- name: RevertData :exec
|
||||
WITH delete_tx AS (
|
||||
DELETE FROM "bitcoin_transactions" WHERE "block_height" >= @from_height
|
||||
RETURNING "tx_hash"
|
||||
), delete_txin AS (
|
||||
DELETE FROM "bitcoin_transaction_txins" WHERE "tx_hash" = ANY(SELECT "tx_hash" FROM delete_tx)
|
||||
RETURNING "prevout_tx_hash", "prevout_tx_idx"
|
||||
), delete_txout AS (
|
||||
DELETE FROM "bitcoin_transaction_txouts" WHERE "tx_hash" = ANY(SELECT "tx_hash" FROM delete_tx)
|
||||
RETURNING "tx_hash", "tx_idx"
|
||||
), revert_txout_spent AS (
|
||||
UPDATE "bitcoin_transaction_txouts"
|
||||
SET "is_spent" = false
|
||||
WHERE
|
||||
("tx_hash", "tx_idx") IN (SELECT "prevout_tx_hash", "prevout_tx_idx" FROM delete_txin) AND
|
||||
("tx_hash", "tx_idx") NOT IN (SELECT "tx_hash", "tx_idx" FROM delete_txout) -- avoid to modified same row twice (modified the same row twice in a single statement is not supported)
|
||||
RETURNING NULL
|
||||
)
|
||||
DELETE FROM "bitcoin_blocks" WHERE "bitcoin_blocks"."block_height" >= @from_height;
|
||||
|
||||
-- name: GetBlockByHeight :one
|
||||
SELECT * FROM bitcoin_blocks WHERE block_height = $1;
|
||||
|
||||
-- name: GetBlocksByHeightRange :many
|
||||
SELECT * FROM bitcoin_blocks WHERE block_height >= @from_height AND block_height <= @to_height ORDER BY block_height ASC;
|
||||
|
||||
-- name: GetTransactionsByHeightRange :many
|
||||
SELECT * FROM bitcoin_transactions WHERE block_height >= @from_height AND block_height <= @to_height;
|
||||
|
||||
-- name: GetTransactionByHash :one
|
||||
SELECT * FROM bitcoin_transactions WHERE tx_hash = $1;
|
||||
|
||||
-- name: GetTransactionTxOutsByTxHashes :many
|
||||
SELECT * FROM bitcoin_transaction_txouts WHERE tx_hash = ANY(@tx_hashes::TEXT[]);
|
||||
|
||||
-- name: GetTransactionTxInsByTxHashes :many
|
||||
SELECT * FROM bitcoin_transaction_txins WHERE tx_hash = ANY(@tx_hashes::TEXT[]);
|
||||
@@ -1,8 +0,0 @@
|
||||
-- name: GetCurrentDBVersion :one
|
||||
SELECT "version" FROM bitcoin_indexer_db_version ORDER BY id DESC LIMIT 1;
|
||||
|
||||
-- name: GetCurrentIndexerStats :one
|
||||
SELECT "client_version", "network" FROM bitcoin_indexer_stats ORDER BY id DESC LIMIT 1;
|
||||
|
||||
-- name: UpdateIndexerStats :exec
|
||||
INSERT INTO bitcoin_indexer_stats (client_version, network) VALUES ($1, $2);
|
||||
@@ -1,25 +0,0 @@
|
||||
package datagateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
)
|
||||
|
||||
type BitcoinDataGateway interface {
|
||||
BitcoinWriterDataDataGateway
|
||||
BitcoinReaderDataDataGateway
|
||||
}
|
||||
|
||||
type BitcoinWriterDataDataGateway interface {
|
||||
InsertBlocks(ctx context.Context, blocks []*types.Block) error
|
||||
RevertBlocks(context.Context, int64) error
|
||||
}
|
||||
|
||||
type BitcoinReaderDataDataGateway interface {
|
||||
GetLatestBlockHeader(context.Context) (types.BlockHeader, error)
|
||||
GetBlockHeaderByHeight(ctx context.Context, blockHeight int64) (types.BlockHeader, error)
|
||||
GetBlocksByHeightRange(ctx context.Context, from int64, to int64) ([]*types.Block, error)
|
||||
GetTransactionByHash(ctx context.Context, txHash chainhash.Hash) (*types.Transaction, error)
|
||||
}
|
||||
@@ -1,13 +0,0 @@
|
||||
package datagateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
)
|
||||
|
||||
type IndexerInformationDataGateway interface {
|
||||
GetCurrentDBVersion(ctx context.Context) (int32, error)
|
||||
GetLatestIndexerStats(ctx context.Context) (version string, network common.Network, err error)
|
||||
UpdateIndexerStats(ctx context.Context, clientVersion string, network common.Network) error
|
||||
}
|
||||
@@ -1,122 +0,0 @@
|
||||
package bitcoin
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/indexers"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/internal/config"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/datagateway"
|
||||
)
|
||||
|
||||
// Make sure to implement the BitcoinProcessor interface
|
||||
var _ indexers.BitcoinProcessor = (*Processor)(nil)
|
||||
|
||||
type Processor struct {
|
||||
config config.Config
|
||||
bitcoinDg datagateway.BitcoinDataGateway
|
||||
indexerInfoDg datagateway.IndexerInformationDataGateway
|
||||
}
|
||||
|
||||
func NewProcessor(config config.Config, bitcoinDg datagateway.BitcoinDataGateway, indexerInfoDg datagateway.IndexerInformationDataGateway) *Processor {
|
||||
return &Processor{
|
||||
config: config,
|
||||
bitcoinDg: bitcoinDg,
|
||||
indexerInfoDg: indexerInfoDg,
|
||||
}
|
||||
}
|
||||
|
||||
func (p Processor) Name() string {
|
||||
return "bitcoin"
|
||||
}
|
||||
|
||||
func (p *Processor) Process(ctx context.Context, inputs []*types.Block) error {
|
||||
if len(inputs) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Process the given blocks before inserting to the database
|
||||
inputs, err := p.process(ctx, inputs)
|
||||
if err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
|
||||
// Insert blocks
|
||||
if err := p.bitcoinDg.InsertBlocks(ctx, inputs); err != nil {
|
||||
return errors.Wrapf(err, "error during insert blocks, from: %d, to: %d", inputs[0].Header.Height, inputs[len(inputs)-1].Header.Height)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Processor) CurrentBlock(ctx context.Context) (types.BlockHeader, error) {
|
||||
b, err := p.bitcoinDg.GetLatestBlockHeader(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.NotFound) {
|
||||
return defaultCurrentBlock, nil
|
||||
}
|
||||
return types.BlockHeader{}, errors.WithStack(err)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func (p *Processor) GetIndexedBlock(ctx context.Context, height int64) (types.BlockHeader, error) {
|
||||
header, err := p.bitcoinDg.GetBlockHeaderByHeight(ctx, height)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.WithStack(err)
|
||||
}
|
||||
return header, nil
|
||||
}
|
||||
|
||||
func (p *Processor) RevertData(ctx context.Context, from int64) error {
|
||||
// to prevent remove txin/txout of duplicated coinbase transaction in the blocks 91842 and 91880
|
||||
// if you really want to revert the data before the block `227835`, you should reset the database and reindex the data instead.
|
||||
if from <= lastV1Block.Height {
|
||||
return errors.Wrapf(errs.InvalidArgument, "can't revert data before block version 2, height: %d", lastV1Block.Height)
|
||||
}
|
||||
|
||||
if err := p.bitcoinDg.RevertBlocks(ctx, from); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Processor) VerifyStates(ctx context.Context) error {
|
||||
// Check current db version with the required db version
|
||||
{
|
||||
dbVersion, err := p.indexerInfoDg.GetCurrentDBVersion(ctx)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "can't get current db version")
|
||||
}
|
||||
|
||||
if dbVersion != DBVersion {
|
||||
return errors.Wrapf(errs.ConflictSetting, "db version mismatch, please upgrade to version %d", DBVersion)
|
||||
}
|
||||
}
|
||||
|
||||
// Check if the latest indexed network is mismatched with configured network
|
||||
{
|
||||
_, network, err := p.indexerInfoDg.GetLatestIndexerStats(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.NotFound) {
|
||||
goto end
|
||||
}
|
||||
return errors.Wrap(err, "can't get latest indexer stats")
|
||||
}
|
||||
|
||||
if network != p.config.Network {
|
||||
return errors.Wrapf(errs.ConflictSetting, "network mismatch, latest indexed network: %q, configured network: %q. If you want to change the network, please reset the database", network, p.config.Network)
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Verify the states of the indexed data to ensure the last shutdown was graceful and no missing data.
|
||||
|
||||
end:
|
||||
if err := p.indexerInfoDg.UpdateIndexerStats(ctx, Version, p.config.Network); err != nil {
|
||||
return errors.Wrap(err, "can't update indexer stats")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -1,91 +0,0 @@
|
||||
package bitcoin
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"slices"
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
)
|
||||
|
||||
// process is a processing rules for the given blocks before inserting to the database
|
||||
//
|
||||
// this function will modify the given data directly.
|
||||
func (p *Processor) process(ctx context.Context, blocks []*types.Block) ([]*types.Block, error) {
|
||||
if len(blocks) == 0 {
|
||||
return blocks, nil
|
||||
}
|
||||
|
||||
// Sort ASC by block height
|
||||
slices.SortFunc(blocks, func(t1, t2 *types.Block) int {
|
||||
return cmp.Compare(t1.Header.Height, t2.Header.Height)
|
||||
})
|
||||
|
||||
if !p.isContinueFromLatestIndexedBlock(ctx, blocks[0]) {
|
||||
return nil, errors.New("given blocks are not continue from the latest indexed block")
|
||||
}
|
||||
|
||||
if !p.isBlocksSequential(blocks) {
|
||||
return nil, errors.New("given blocks are not in sequence")
|
||||
}
|
||||
|
||||
p.removeDuplicateCoinbaseTxInputsOutputs(blocks)
|
||||
|
||||
return blocks, nil
|
||||
}
|
||||
|
||||
// check if the given blocks are continue from the latest indexed block
|
||||
// to prevent inserting out-of-order blocks or duplicate blocks
|
||||
func (p *Processor) isBlocksSequential(blocks []*types.Block) bool {
|
||||
if len(blocks) == 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
for i, block := range blocks {
|
||||
if i == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
if block.Header.Height != blocks[i-1].Header.Height+1 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// check if the given blocks are continue from the latest indexed block
|
||||
// to prevent inserting out-of-order blocks or duplicate blocks
|
||||
func (p *Processor) isContinueFromLatestIndexedBlock(ctx context.Context, block *types.Block) bool {
|
||||
latestBlock, err := p.CurrentBlock(ctx)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
return block.Header.Height == latestBlock.Height+1
|
||||
}
|
||||
|
||||
// there 2 coinbase transaction that are duplicated in the blocks 91842 and 91880.
|
||||
// if the given block version is v1 and height is `91842` or `91880`,
|
||||
// then remove transaction inputs/outputs to prevent duplicate txin/txout error when inserting to the database.
|
||||
//
|
||||
// Theses duplicated coinbase transactions are having the same transaction input/output and
|
||||
// utxo from these 2 duplicated coinbase txs can redeem only once. so, it's safe to remove them and can
|
||||
// use inputs/outputs from the previous block.
|
||||
//
|
||||
// Duplicate Coinbase Transactions:
|
||||
// - `454279874213763724535987336644243549a273058910332236515429488599` in blocks 91812, 91842
|
||||
// - `e3bf3d07d4b0375638d5f1db5255fe07ba2c4cb067cd81b84ee974b6585fb468` in blocks 91722, 91880
|
||||
//
|
||||
// This function will modify the given data directly.
|
||||
func (p *Processor) removeDuplicateCoinbaseTxInputsOutputs(blocks []*types.Block) {
|
||||
for _, block := range blocks {
|
||||
header := block.Header
|
||||
if header.Version == 1 && (header.Height == 91842 || header.Height == 91880) {
|
||||
// remove transaction inputs/outputs from coinbase transaction (first transaction)
|
||||
block.Transactions[0].TxIn = nil
|
||||
block.Transactions[0].TxOut = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,144 +0,0 @@
|
||||
package bitcoin
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestDuplicateCoinbaseTxHashHandling(t *testing.T) {
|
||||
processor := Processor{}
|
||||
generator := func() []*types.Block {
|
||||
return []*types.Block{
|
||||
{
|
||||
Header: types.BlockHeader{Height: 91842, Version: 1},
|
||||
Transactions: []*types.Transaction{
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Header: types.BlockHeader{Height: 91880, Version: 1},
|
||||
Transactions: []*types.Transaction{
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("all_duplicated_txs", func(t *testing.T) {
|
||||
blocks := generator()
|
||||
processor.removeDuplicateCoinbaseTxInputsOutputs(blocks)
|
||||
|
||||
assert.Len(t, blocks, 2, "should not remove any blocks")
|
||||
for _, block := range blocks {
|
||||
assert.Len(t, block.Transactions, 2, "should not remove any transactions")
|
||||
assert.Len(t, block.Transactions[0].TxIn, 0, "should remove tx inputs from coinbase transaction")
|
||||
assert.Len(t, block.Transactions[0].TxOut, 0, "should remove tx outputs from coinbase transaction")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("not_duplicated_txs", func(t *testing.T) {
|
||||
blocks := []*types.Block{
|
||||
{
|
||||
Header: types.BlockHeader{Height: 91812, Version: 1},
|
||||
Transactions: []*types.Transaction{
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Header: types.BlockHeader{Height: 91722, Version: 1},
|
||||
Transactions: []*types.Transaction{
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
processor.removeDuplicateCoinbaseTxInputsOutputs(blocks)
|
||||
|
||||
assert.Len(t, blocks, 2, "should not remove any blocks")
|
||||
for _, block := range blocks {
|
||||
assert.Len(t, block.Transactions, 2, "should not remove any transactions")
|
||||
assert.Len(t, block.Transactions[0].TxIn, 4, "should not remove tx inputs from coinbase transaction")
|
||||
assert.Len(t, block.Transactions[0].TxOut, 4, "should not remove tx outputs from coinbase transaction")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("mixed", func(t *testing.T) {
|
||||
blocks := []*types.Block{
|
||||
{
|
||||
Header: types.BlockHeader{Height: 91812, Version: 1},
|
||||
Transactions: []*types.Transaction{
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
blocks = append(blocks, generator()...)
|
||||
blocks = append(blocks, &types.Block{
|
||||
Header: types.BlockHeader{Height: 91722, Version: 1},
|
||||
Transactions: []*types.Transaction{
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
{
|
||||
TxIn: []*types.TxIn{{}, {}, {}, {}},
|
||||
TxOut: []*types.TxOut{{}, {}, {}, {}},
|
||||
},
|
||||
},
|
||||
})
|
||||
processor.removeDuplicateCoinbaseTxInputsOutputs(blocks)
|
||||
|
||||
assert.Len(t, blocks, 4, "should not remove any blocks")
|
||||
|
||||
// only 2nd and 3rd blocks should be modified
|
||||
for i, block := range blocks {
|
||||
t.Run(fmt.Sprint(i), func(t *testing.T) {
|
||||
if i == 1 || i == 2 {
|
||||
assert.Len(t, block.Transactions, 2, "should not remove any transactions")
|
||||
assert.Len(t, block.Transactions[0].TxIn, 0, "should remove tx inputs from coinbase transaction")
|
||||
assert.Len(t, block.Transactions[0].TxOut, 0, "should remove tx outputs from coinbase transaction")
|
||||
} else {
|
||||
assert.Len(t, block.Transactions, 2, "should not remove any transactions")
|
||||
assert.Lenf(t, block.Transactions[0].TxIn, 4, "should not remove tx inputs from coinbase transaction")
|
||||
assert.Len(t, block.Transactions[0].TxOut, 4, "should not remove tx outputs from coinbase transaction")
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -1,169 +0,0 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/repository/postgres/gen"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/samber/lo"
|
||||
)
|
||||
|
||||
func (r *Repository) GetLatestBlockHeader(ctx context.Context) (types.BlockHeader, error) {
|
||||
model, err := r.queries.GetLatestBlockHeader(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return types.BlockHeader{}, errors.Join(errs.NotFound, err)
|
||||
}
|
||||
return types.BlockHeader{}, errors.Wrap(err, "failed to get latest block header")
|
||||
}
|
||||
|
||||
data, err := mapBlockHeaderModelToType(model)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Wrap(err, "failed to map block header model to type")
|
||||
}
|
||||
|
||||
return data, nil
|
||||
}
|
||||
|
||||
func (r *Repository) InsertBlocks(ctx context.Context, blocks []*types.Block) error {
|
||||
if len(blocks) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
blockParams, txParams, txoutParams, txinParams := mapBlocksTypeToParams(blocks)
|
||||
|
||||
tx, err := r.db.Begin(ctx)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to begin transaction")
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
queries := r.queries.WithTx(tx)
|
||||
|
||||
if err := queries.BatchInsertBlocks(ctx, blockParams); err != nil {
|
||||
return errors.Wrap(err, "failed to batch insert block headers")
|
||||
}
|
||||
|
||||
if err := queries.BatchInsertTransactions(ctx, txParams); err != nil {
|
||||
return errors.Wrap(err, "failed to batch insert transactions")
|
||||
}
|
||||
|
||||
// Should insert txout first, then txin
|
||||
// Because txin references txout
|
||||
if err := queries.BatchInsertTransactionTxOuts(ctx, txoutParams); err != nil {
|
||||
return errors.Wrap(err, "failed to batch insert transaction txins")
|
||||
}
|
||||
|
||||
if err := queries.BatchInsertTransactionTxIns(ctx, txinParams); err != nil {
|
||||
return errors.Wrap(err, "failed to batch insert transaction txins")
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return errors.Wrap(err, "failed to commit transaction")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Repository) RevertBlocks(ctx context.Context, from int64) error {
|
||||
tx, err := r.db.Begin(ctx)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to begin transaction")
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
queries := r.queries.WithTx(tx)
|
||||
if err := queries.RevertData(ctx, int32(from)); err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return errors.Wrap(err, "failed to revert data")
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return errors.Wrap(err, "failed to commit transaction")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Repository) GetBlockHeaderByHeight(ctx context.Context, blockHeight int64) (types.BlockHeader, error) {
|
||||
blockModel, err := r.queries.GetBlockByHeight(ctx, int32(blockHeight))
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return types.BlockHeader{}, errors.Join(errs.NotFound, err)
|
||||
}
|
||||
return types.BlockHeader{}, errors.Wrap(err, "failed to get block by height")
|
||||
}
|
||||
|
||||
data, err := mapBlockHeaderModelToType(blockModel)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Wrap(err, "failed to map block header model to type")
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
func (r *Repository) GetBlocksByHeightRange(ctx context.Context, from int64, to int64) ([]*types.Block, error) {
|
||||
blocks, err := r.queries.GetBlocksByHeightRange(ctx, gen.GetBlocksByHeightRangeParams{
|
||||
FromHeight: int32(from),
|
||||
ToHeight: int32(to),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to get blocks by height range")
|
||||
}
|
||||
|
||||
if len(blocks) == 0 {
|
||||
return []*types.Block{}, nil
|
||||
}
|
||||
|
||||
txs, err := r.queries.GetTransactionsByHeightRange(ctx, gen.GetTransactionsByHeightRangeParams{
|
||||
FromHeight: int32(from),
|
||||
ToHeight: int32(to),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to get transactions by height range")
|
||||
}
|
||||
|
||||
txHashes := lo.Map(txs, func(tx gen.BitcoinTransaction, _ int) string { return tx.TxHash })
|
||||
|
||||
txOuts, err := r.queries.GetTransactionTxOutsByTxHashes(ctx, txHashes)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to get transaction txouts by tx hashes")
|
||||
}
|
||||
|
||||
txIns, err := r.queries.GetTransactionTxInsByTxHashes(ctx, txHashes)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to get transaction txins by tx hashes")
|
||||
}
|
||||
|
||||
// Grouping result by block height and tx hash
|
||||
groupedTxs := lo.GroupBy(txs, func(tx gen.BitcoinTransaction) int32 { return tx.BlockHeight })
|
||||
groupedTxOuts := lo.GroupBy(txOuts, func(txOut gen.BitcoinTransactionTxout) string { return txOut.TxHash })
|
||||
groupedTxIns := lo.GroupBy(txIns, func(txIn gen.BitcoinTransactionTxin) string { return txIn.TxHash })
|
||||
|
||||
var errs []error
|
||||
result := lo.Map(blocks, func(blockModel gen.BitcoinBlock, _ int) *types.Block {
|
||||
header, err := mapBlockHeaderModelToType(blockModel)
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrap(err, "failed to map block header model to type"))
|
||||
return nil
|
||||
}
|
||||
|
||||
txsModel := groupedTxs[blockModel.BlockHeight]
|
||||
return &types.Block{
|
||||
Header: header,
|
||||
Transactions: lo.Map(txsModel, func(txModel gen.BitcoinTransaction, _ int) *types.Transaction {
|
||||
tx, err := mapTransactionModelToType(txModel, groupedTxIns[txModel.TxHash], groupedTxOuts[txModel.TxHash])
|
||||
if err != nil {
|
||||
errs = append(errs, errors.Wrap(err, "failed to map transaction model to type"))
|
||||
return nil
|
||||
}
|
||||
return &tx
|
||||
}),
|
||||
}
|
||||
})
|
||||
if len(errs) > 0 {
|
||||
return nil, errors.Wrap(errors.Join(errs...), "failed while mapping result")
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -1,408 +0,0 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.26.0
|
||||
// source: data.sql
|
||||
|
||||
package gen
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
const batchInsertBlocks = `-- name: BatchInsertBlocks :exec
|
||||
INSERT INTO bitcoin_blocks ("block_height","block_hash","version","merkle_root","prev_block_hash","timestamp","bits","nonce")
|
||||
VALUES (
|
||||
unnest($1::INT[]),
|
||||
unnest($2::TEXT[]),
|
||||
unnest($3::INT[]),
|
||||
unnest($4::TEXT[]),
|
||||
unnest($5::TEXT[]),
|
||||
unnest($6::TIMESTAMP WITH TIME ZONE[]), -- or use TIMESTAMPTZ
|
||||
unnest($7::BIGINT[]),
|
||||
unnest($8::BIGINT[])
|
||||
)
|
||||
`
|
||||
|
||||
type BatchInsertBlocksParams struct {
|
||||
BlockHeightArr []int32
|
||||
BlockHashArr []string
|
||||
VersionArr []int32
|
||||
MerkleRootArr []string
|
||||
PrevBlockHashArr []string
|
||||
TimestampArr []pgtype.Timestamptz
|
||||
BitsArr []int64
|
||||
NonceArr []int64
|
||||
}
|
||||
|
||||
func (q *Queries) BatchInsertBlocks(ctx context.Context, arg BatchInsertBlocksParams) error {
|
||||
_, err := q.db.Exec(ctx, batchInsertBlocks,
|
||||
arg.BlockHeightArr,
|
||||
arg.BlockHashArr,
|
||||
arg.VersionArr,
|
||||
arg.MerkleRootArr,
|
||||
arg.PrevBlockHashArr,
|
||||
arg.TimestampArr,
|
||||
arg.BitsArr,
|
||||
arg.NonceArr,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const batchInsertTransactionTxIns = `-- name: BatchInsertTransactionTxIns :exec
|
||||
WITH update_txout AS (
|
||||
UPDATE "bitcoin_transaction_txouts"
|
||||
SET "is_spent" = true
|
||||
FROM (SELECT unnest($1::TEXT[]) as tx_hash, unnest($2::INT[]) as tx_idx) as txin
|
||||
WHERE "bitcoin_transaction_txouts"."tx_hash" = txin.tx_hash AND "bitcoin_transaction_txouts"."tx_idx" = txin.tx_idx AND "is_spent" = false
|
||||
RETURNING "bitcoin_transaction_txouts"."tx_hash", "bitcoin_transaction_txouts"."tx_idx", "pkscript"
|
||||
), prepare_insert AS (
|
||||
SELECT input.tx_hash, input.tx_idx, prevout_tx_hash, prevout_tx_idx, update_txout.pkscript as prevout_pkscript, scriptsig, witness, sequence
|
||||
FROM (
|
||||
SELECT
|
||||
unnest($3::TEXT[]) as tx_hash,
|
||||
unnest($4::INT[]) as tx_idx,
|
||||
unnest($1::TEXT[]) as prevout_tx_hash,
|
||||
unnest($2::INT[]) as prevout_tx_idx,
|
||||
unnest($5::TEXT[]) as scriptsig,
|
||||
unnest($6::TEXT[]) as witness,
|
||||
unnest($7::INT[]) as sequence
|
||||
) input LEFT JOIN update_txout ON "update_txout"."tx_hash" = "input"."prevout_tx_hash" AND "update_txout"."tx_idx" = "input"."prevout_tx_idx"
|
||||
)
|
||||
INSERT INTO bitcoin_transaction_txins ("tx_hash","tx_idx","prevout_tx_hash","prevout_tx_idx", "prevout_pkscript","scriptsig","witness","sequence")
|
||||
SELECT "tx_hash", "tx_idx", "prevout_tx_hash", "prevout_tx_idx", "prevout_pkscript", "scriptsig", "witness", "sequence" FROM prepare_insert
|
||||
`
|
||||
|
||||
type BatchInsertTransactionTxInsParams struct {
|
||||
PrevoutTxHashArr []string
|
||||
PrevoutTxIdxArr []int32
|
||||
TxHashArr []string
|
||||
TxIdxArr []int32
|
||||
ScriptsigArr []string
|
||||
WitnessArr []string
|
||||
SequenceArr []int32
|
||||
}
|
||||
|
||||
func (q *Queries) BatchInsertTransactionTxIns(ctx context.Context, arg BatchInsertTransactionTxInsParams) error {
|
||||
_, err := q.db.Exec(ctx, batchInsertTransactionTxIns,
|
||||
arg.PrevoutTxHashArr,
|
||||
arg.PrevoutTxIdxArr,
|
||||
arg.TxHashArr,
|
||||
arg.TxIdxArr,
|
||||
arg.ScriptsigArr,
|
||||
arg.WitnessArr,
|
||||
arg.SequenceArr,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const batchInsertTransactionTxOuts = `-- name: BatchInsertTransactionTxOuts :exec
|
||||
INSERT INTO bitcoin_transaction_txouts ("tx_hash","tx_idx","pkscript","value")
|
||||
VALUES (
|
||||
unnest($1::TEXT[]),
|
||||
unnest($2::INT[]),
|
||||
unnest($3::TEXT[]),
|
||||
unnest($4::BIGINT[])
|
||||
)
|
||||
`
|
||||
|
||||
type BatchInsertTransactionTxOutsParams struct {
|
||||
TxHashArr []string
|
||||
TxIdxArr []int32
|
||||
PkscriptArr []string
|
||||
ValueArr []int64
|
||||
}
|
||||
|
||||
func (q *Queries) BatchInsertTransactionTxOuts(ctx context.Context, arg BatchInsertTransactionTxOutsParams) error {
|
||||
_, err := q.db.Exec(ctx, batchInsertTransactionTxOuts,
|
||||
arg.TxHashArr,
|
||||
arg.TxIdxArr,
|
||||
arg.PkscriptArr,
|
||||
arg.ValueArr,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const batchInsertTransactions = `-- name: BatchInsertTransactions :exec
|
||||
INSERT INTO bitcoin_transactions ("tx_hash","version","locktime","block_height","block_hash","idx")
|
||||
VALUES (
|
||||
unnest($1::TEXT[]),
|
||||
unnest($2::INT[]),
|
||||
unnest($3::BIGINT[]),
|
||||
unnest($4::INT[]),
|
||||
unnest($5::TEXT[]),
|
||||
unnest($6::INT[])
|
||||
)
|
||||
`
|
||||
|
||||
type BatchInsertTransactionsParams struct {
|
||||
TxHashArr []string
|
||||
VersionArr []int32
|
||||
LocktimeArr []int64
|
||||
BlockHeightArr []int32
|
||||
BlockHashArr []string
|
||||
IdxArr []int32
|
||||
}
|
||||
|
||||
func (q *Queries) BatchInsertTransactions(ctx context.Context, arg BatchInsertTransactionsParams) error {
|
||||
_, err := q.db.Exec(ctx, batchInsertTransactions,
|
||||
arg.TxHashArr,
|
||||
arg.VersionArr,
|
||||
arg.LocktimeArr,
|
||||
arg.BlockHeightArr,
|
||||
arg.BlockHashArr,
|
||||
arg.IdxArr,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const getBlockByHeight = `-- name: GetBlockByHeight :one
|
||||
SELECT block_height, block_hash, version, merkle_root, prev_block_hash, timestamp, bits, nonce FROM bitcoin_blocks WHERE block_height = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetBlockByHeight(ctx context.Context, blockHeight int32) (BitcoinBlock, error) {
|
||||
row := q.db.QueryRow(ctx, getBlockByHeight, blockHeight)
|
||||
var i BitcoinBlock
|
||||
err := row.Scan(
|
||||
&i.BlockHeight,
|
||||
&i.BlockHash,
|
||||
&i.Version,
|
||||
&i.MerkleRoot,
|
||||
&i.PrevBlockHash,
|
||||
&i.Timestamp,
|
||||
&i.Bits,
|
||||
&i.Nonce,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getBlocksByHeightRange = `-- name: GetBlocksByHeightRange :many
|
||||
SELECT block_height, block_hash, version, merkle_root, prev_block_hash, timestamp, bits, nonce FROM bitcoin_blocks WHERE block_height >= $1 AND block_height <= $2 ORDER BY block_height ASC
|
||||
`
|
||||
|
||||
type GetBlocksByHeightRangeParams struct {
|
||||
FromHeight int32
|
||||
ToHeight int32
|
||||
}
|
||||
|
||||
func (q *Queries) GetBlocksByHeightRange(ctx context.Context, arg GetBlocksByHeightRangeParams) ([]BitcoinBlock, error) {
|
||||
rows, err := q.db.Query(ctx, getBlocksByHeightRange, arg.FromHeight, arg.ToHeight)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []BitcoinBlock
|
||||
for rows.Next() {
|
||||
var i BitcoinBlock
|
||||
if err := rows.Scan(
|
||||
&i.BlockHeight,
|
||||
&i.BlockHash,
|
||||
&i.Version,
|
||||
&i.MerkleRoot,
|
||||
&i.PrevBlockHash,
|
||||
&i.Timestamp,
|
||||
&i.Bits,
|
||||
&i.Nonce,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const getLatestBlockHeader = `-- name: GetLatestBlockHeader :one
|
||||
SELECT block_height, block_hash, version, merkle_root, prev_block_hash, timestamp, bits, nonce FROM bitcoin_blocks ORDER BY block_height DESC LIMIT 1
|
||||
`
|
||||
|
||||
func (q *Queries) GetLatestBlockHeader(ctx context.Context) (BitcoinBlock, error) {
|
||||
row := q.db.QueryRow(ctx, getLatestBlockHeader)
|
||||
var i BitcoinBlock
|
||||
err := row.Scan(
|
||||
&i.BlockHeight,
|
||||
&i.BlockHash,
|
||||
&i.Version,
|
||||
&i.MerkleRoot,
|
||||
&i.PrevBlockHash,
|
||||
&i.Timestamp,
|
||||
&i.Bits,
|
||||
&i.Nonce,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getTransactionByHash = `-- name: GetTransactionByHash :one
|
||||
SELECT tx_hash, version, locktime, block_height, block_hash, idx FROM bitcoin_transactions WHERE tx_hash = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetTransactionByHash(ctx context.Context, txHash string) (BitcoinTransaction, error) {
|
||||
row := q.db.QueryRow(ctx, getTransactionByHash, txHash)
|
||||
var i BitcoinTransaction
|
||||
err := row.Scan(
|
||||
&i.TxHash,
|
||||
&i.Version,
|
||||
&i.Locktime,
|
||||
&i.BlockHeight,
|
||||
&i.BlockHash,
|
||||
&i.Idx,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getTransactionTxInsByTxHashes = `-- name: GetTransactionTxInsByTxHashes :many
|
||||
SELECT tx_hash, tx_idx, prevout_tx_hash, prevout_tx_idx, prevout_pkscript, scriptsig, witness, sequence FROM bitcoin_transaction_txins WHERE tx_hash = ANY($1::TEXT[])
|
||||
`
|
||||
|
||||
func (q *Queries) GetTransactionTxInsByTxHashes(ctx context.Context, txHashes []string) ([]BitcoinTransactionTxin, error) {
|
||||
rows, err := q.db.Query(ctx, getTransactionTxInsByTxHashes, txHashes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []BitcoinTransactionTxin
|
||||
for rows.Next() {
|
||||
var i BitcoinTransactionTxin
|
||||
if err := rows.Scan(
|
||||
&i.TxHash,
|
||||
&i.TxIdx,
|
||||
&i.PrevoutTxHash,
|
||||
&i.PrevoutTxIdx,
|
||||
&i.PrevoutPkscript,
|
||||
&i.Scriptsig,
|
||||
&i.Witness,
|
||||
&i.Sequence,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const getTransactionTxOutsByTxHashes = `-- name: GetTransactionTxOutsByTxHashes :many
|
||||
SELECT tx_hash, tx_idx, pkscript, value, is_spent FROM bitcoin_transaction_txouts WHERE tx_hash = ANY($1::TEXT[])
|
||||
`
|
||||
|
||||
func (q *Queries) GetTransactionTxOutsByTxHashes(ctx context.Context, txHashes []string) ([]BitcoinTransactionTxout, error) {
|
||||
rows, err := q.db.Query(ctx, getTransactionTxOutsByTxHashes, txHashes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []BitcoinTransactionTxout
|
||||
for rows.Next() {
|
||||
var i BitcoinTransactionTxout
|
||||
if err := rows.Scan(
|
||||
&i.TxHash,
|
||||
&i.TxIdx,
|
||||
&i.Pkscript,
|
||||
&i.Value,
|
||||
&i.IsSpent,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const getTransactionsByHeightRange = `-- name: GetTransactionsByHeightRange :many
|
||||
SELECT tx_hash, version, locktime, block_height, block_hash, idx FROM bitcoin_transactions WHERE block_height >= $1 AND block_height <= $2
|
||||
`
|
||||
|
||||
type GetTransactionsByHeightRangeParams struct {
|
||||
FromHeight int32
|
||||
ToHeight int32
|
||||
}
|
||||
|
||||
func (q *Queries) GetTransactionsByHeightRange(ctx context.Context, arg GetTransactionsByHeightRangeParams) ([]BitcoinTransaction, error) {
|
||||
rows, err := q.db.Query(ctx, getTransactionsByHeightRange, arg.FromHeight, arg.ToHeight)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []BitcoinTransaction
|
||||
for rows.Next() {
|
||||
var i BitcoinTransaction
|
||||
if err := rows.Scan(
|
||||
&i.TxHash,
|
||||
&i.Version,
|
||||
&i.Locktime,
|
||||
&i.BlockHeight,
|
||||
&i.BlockHash,
|
||||
&i.Idx,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const insertBlock = `-- name: InsertBlock :exec
|
||||
INSERT INTO bitcoin_blocks ("block_height","block_hash","version","merkle_root","prev_block_hash","timestamp","bits","nonce") VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
||||
`
|
||||
|
||||
type InsertBlockParams struct {
|
||||
BlockHeight int32
|
||||
BlockHash string
|
||||
Version int32
|
||||
MerkleRoot string
|
||||
PrevBlockHash string
|
||||
Timestamp pgtype.Timestamptz
|
||||
Bits int64
|
||||
Nonce int64
|
||||
}
|
||||
|
||||
func (q *Queries) InsertBlock(ctx context.Context, arg InsertBlockParams) error {
|
||||
_, err := q.db.Exec(ctx, insertBlock,
|
||||
arg.BlockHeight,
|
||||
arg.BlockHash,
|
||||
arg.Version,
|
||||
arg.MerkleRoot,
|
||||
arg.PrevBlockHash,
|
||||
arg.Timestamp,
|
||||
arg.Bits,
|
||||
arg.Nonce,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const revertData = `-- name: RevertData :exec
|
||||
WITH delete_tx AS (
|
||||
DELETE FROM "bitcoin_transactions" WHERE "block_height" >= $1
|
||||
RETURNING "tx_hash"
|
||||
), delete_txin AS (
|
||||
DELETE FROM "bitcoin_transaction_txins" WHERE "tx_hash" = ANY(SELECT "tx_hash" FROM delete_tx)
|
||||
RETURNING "prevout_tx_hash", "prevout_tx_idx"
|
||||
), delete_txout AS (
|
||||
DELETE FROM "bitcoin_transaction_txouts" WHERE "tx_hash" = ANY(SELECT "tx_hash" FROM delete_tx)
|
||||
RETURNING "tx_hash", "tx_idx"
|
||||
), revert_txout_spent AS (
|
||||
UPDATE "bitcoin_transaction_txouts"
|
||||
SET "is_spent" = false
|
||||
WHERE
|
||||
("tx_hash", "tx_idx") IN (SELECT "prevout_tx_hash", "prevout_tx_idx" FROM delete_txin) AND
|
||||
("tx_hash", "tx_idx") NOT IN (SELECT "tx_hash", "tx_idx" FROM delete_txout) -- avoid to modified same row twice (modified the same row twice in a single statement is not supported)
|
||||
RETURNING NULL
|
||||
)
|
||||
DELETE FROM "bitcoin_blocks" WHERE "bitcoin_blocks"."block_height" >= $1
|
||||
`
|
||||
|
||||
func (q *Queries) RevertData(ctx context.Context, fromHeight int32) error {
|
||||
_, err := q.db.Exec(ctx, revertData, fromHeight)
|
||||
return err
|
||||
}
|
||||
@@ -1,32 +0,0 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.26.0
|
||||
|
||||
package gen
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
type DBTX interface {
|
||||
Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
|
||||
Query(context.Context, string, ...interface{}) (pgx.Rows, error)
|
||||
QueryRow(context.Context, string, ...interface{}) pgx.Row
|
||||
}
|
||||
|
||||
func New(db DBTX) *Queries {
|
||||
return &Queries{db: db}
|
||||
}
|
||||
|
||||
type Queries struct {
|
||||
db DBTX
|
||||
}
|
||||
|
||||
func (q *Queries) WithTx(tx pgx.Tx) *Queries {
|
||||
return &Queries{
|
||||
db: tx,
|
||||
}
|
||||
}
|
||||
@@ -1,51 +0,0 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.26.0
|
||||
// source: info.sql
|
||||
|
||||
package gen
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
const getCurrentDBVersion = `-- name: GetCurrentDBVersion :one
|
||||
SELECT "version" FROM bitcoin_indexer_db_version ORDER BY id DESC LIMIT 1
|
||||
`
|
||||
|
||||
func (q *Queries) GetCurrentDBVersion(ctx context.Context) (int32, error) {
|
||||
row := q.db.QueryRow(ctx, getCurrentDBVersion)
|
||||
var version int32
|
||||
err := row.Scan(&version)
|
||||
return version, err
|
||||
}
|
||||
|
||||
const getCurrentIndexerStats = `-- name: GetCurrentIndexerStats :one
|
||||
SELECT "client_version", "network" FROM bitcoin_indexer_stats ORDER BY id DESC LIMIT 1
|
||||
`
|
||||
|
||||
type GetCurrentIndexerStatsRow struct {
|
||||
ClientVersion string
|
||||
Network string
|
||||
}
|
||||
|
||||
func (q *Queries) GetCurrentIndexerStats(ctx context.Context) (GetCurrentIndexerStatsRow, error) {
|
||||
row := q.db.QueryRow(ctx, getCurrentIndexerStats)
|
||||
var i GetCurrentIndexerStatsRow
|
||||
err := row.Scan(&i.ClientVersion, &i.Network)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const updateIndexerStats = `-- name: UpdateIndexerStats :exec
|
||||
INSERT INTO bitcoin_indexer_stats (client_version, network) VALUES ($1, $2)
|
||||
`
|
||||
|
||||
type UpdateIndexerStatsParams struct {
|
||||
ClientVersion string
|
||||
Network string
|
||||
}
|
||||
|
||||
func (q *Queries) UpdateIndexerStats(ctx context.Context, arg UpdateIndexerStatsParams) error {
|
||||
_, err := q.db.Exec(ctx, updateIndexerStats, arg.ClientVersion, arg.Network)
|
||||
return err
|
||||
}
|
||||
@@ -1,61 +0,0 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.26.0
|
||||
|
||||
package gen
|
||||
|
||||
import (
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
type BitcoinBlock struct {
|
||||
BlockHeight int32
|
||||
BlockHash string
|
||||
Version int32
|
||||
MerkleRoot string
|
||||
PrevBlockHash string
|
||||
Timestamp pgtype.Timestamptz
|
||||
Bits int64
|
||||
Nonce int64
|
||||
}
|
||||
|
||||
type BitcoinIndexerDbVersion struct {
|
||||
Id int64
|
||||
Version int32
|
||||
CreatedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
type BitcoinIndexerStat struct {
|
||||
Id int64
|
||||
ClientVersion string
|
||||
Network string
|
||||
CreatedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
type BitcoinTransaction struct {
|
||||
TxHash string
|
||||
Version int32
|
||||
Locktime int64
|
||||
BlockHeight int32
|
||||
BlockHash string
|
||||
Idx int32
|
||||
}
|
||||
|
||||
type BitcoinTransactionTxin struct {
|
||||
TxHash string
|
||||
TxIdx int32
|
||||
PrevoutTxHash string
|
||||
PrevoutTxIdx int32
|
||||
PrevoutPkscript pgtype.Text
|
||||
Scriptsig string
|
||||
Witness string
|
||||
Sequence int64
|
||||
}
|
||||
|
||||
type BitcoinTransactionTxout struct {
|
||||
TxHash string
|
||||
TxIdx int32
|
||||
Pkscript string
|
||||
Value int64
|
||||
IsSpent bool
|
||||
}
|
||||
@@ -1,44 +0,0 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/datagateway"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/repository/postgres/gen"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// Make sure Repository implements the IndexerInformationDataGateway interface
|
||||
var _ datagateway.IndexerInformationDataGateway = (*Repository)(nil)
|
||||
|
||||
func (r *Repository) GetCurrentDBVersion(ctx context.Context) (int32, error) {
|
||||
version, err := r.queries.GetCurrentDBVersion(ctx)
|
||||
if err != nil {
|
||||
return 0, errors.WithStack(err)
|
||||
}
|
||||
return version, nil
|
||||
}
|
||||
|
||||
func (r *Repository) GetLatestIndexerStats(ctx context.Context) (string, common.Network, error) {
|
||||
stats, err := r.queries.GetCurrentIndexerStats(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return "", "", errors.Join(errs.NotFound, err)
|
||||
}
|
||||
return "", "", errors.WithStack(err)
|
||||
}
|
||||
return stats.ClientVersion, common.Network(stats.Network), nil
|
||||
}
|
||||
|
||||
func (r *Repository) UpdateIndexerStats(ctx context.Context, clientVersion string, network common.Network) error {
|
||||
if err := r.queries.UpdateIndexerStats(ctx, gen.UpdateIndexerStatsParams{
|
||||
ClientVersion: clientVersion,
|
||||
Network: network.String(),
|
||||
}); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,197 +0,0 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"encoding/hex"
|
||||
"slices"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/repository/postgres/gen"
|
||||
"github.com/gaze-network/indexer-network/pkg/btcutils"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
func mapBlockHeaderModelToType(src gen.BitcoinBlock) (types.BlockHeader, error) {
|
||||
hash, err := chainhash.NewHashFromStr(src.BlockHash)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Join(errors.Wrap(err, "failed to parse block hash"), errs.InternalError)
|
||||
}
|
||||
prevHash, err := chainhash.NewHashFromStr(src.PrevBlockHash)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Join(errors.Wrap(err, "failed to parse prev block hash"), errs.InternalError)
|
||||
}
|
||||
merkleRoot, err := chainhash.NewHashFromStr(src.MerkleRoot)
|
||||
if err != nil {
|
||||
return types.BlockHeader{}, errors.Join(errors.Wrap(err, "failed to parse merkle root"), errs.InternalError)
|
||||
}
|
||||
return types.BlockHeader{
|
||||
Hash: *hash,
|
||||
Height: int64(src.BlockHeight),
|
||||
Version: src.Version,
|
||||
PrevBlock: *prevHash,
|
||||
MerkleRoot: *merkleRoot,
|
||||
Timestamp: src.Timestamp.Time,
|
||||
Bits: uint32(src.Bits),
|
||||
Nonce: uint32(src.Nonce),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func mapBlocksTypeToParams(src []*types.Block) (gen.BatchInsertBlocksParams, gen.BatchInsertTransactionsParams, gen.BatchInsertTransactionTxOutsParams, gen.BatchInsertTransactionTxInsParams) {
|
||||
blocks := gen.BatchInsertBlocksParams{
|
||||
BlockHeightArr: make([]int32, 0, len(src)),
|
||||
BlockHashArr: make([]string, 0, len(src)),
|
||||
VersionArr: make([]int32, 0, len(src)),
|
||||
MerkleRootArr: make([]string, 0, len(src)),
|
||||
PrevBlockHashArr: make([]string, 0, len(src)),
|
||||
TimestampArr: make([]pgtype.Timestamptz, 0, len(src)),
|
||||
BitsArr: make([]int64, 0, len(src)),
|
||||
NonceArr: make([]int64, 0, len(src)),
|
||||
}
|
||||
txs := gen.BatchInsertTransactionsParams{
|
||||
TxHashArr: []string{},
|
||||
VersionArr: []int32{},
|
||||
LocktimeArr: []int64{},
|
||||
BlockHeightArr: []int32{},
|
||||
BlockHashArr: []string{},
|
||||
IdxArr: []int32{},
|
||||
}
|
||||
txouts := gen.BatchInsertTransactionTxOutsParams{
|
||||
TxHashArr: []string{},
|
||||
TxIdxArr: []int32{},
|
||||
PkscriptArr: []string{},
|
||||
ValueArr: []int64{},
|
||||
}
|
||||
txins := gen.BatchInsertTransactionTxInsParams{
|
||||
PrevoutTxHashArr: []string{},
|
||||
PrevoutTxIdxArr: []int32{},
|
||||
TxHashArr: []string{},
|
||||
TxIdxArr: []int32{},
|
||||
ScriptsigArr: []string{},
|
||||
WitnessArr: []string{},
|
||||
SequenceArr: []int32{},
|
||||
}
|
||||
|
||||
for _, block := range src {
|
||||
blockHash := block.Header.Hash.String()
|
||||
|
||||
// Batch insert blocks
|
||||
blocks.BlockHeightArr = append(blocks.BlockHeightArr, int32(block.Header.Height))
|
||||
blocks.BlockHashArr = append(blocks.BlockHashArr, blockHash)
|
||||
blocks.VersionArr = append(blocks.VersionArr, block.Header.Version)
|
||||
blocks.MerkleRootArr = append(blocks.MerkleRootArr, block.Header.MerkleRoot.String())
|
||||
blocks.PrevBlockHashArr = append(blocks.PrevBlockHashArr, block.Header.PrevBlock.String())
|
||||
blocks.TimestampArr = append(blocks.TimestampArr, pgtype.Timestamptz{
|
||||
Time: block.Header.Timestamp,
|
||||
Valid: true,
|
||||
})
|
||||
blocks.BitsArr = append(blocks.BitsArr, int64(block.Header.Bits))
|
||||
blocks.NonceArr = append(blocks.NonceArr, int64(block.Header.Nonce))
|
||||
|
||||
for txIdx, srcTx := range block.Transactions {
|
||||
txHash := srcTx.TxHash.String()
|
||||
|
||||
// Batch insert transactions
|
||||
txs.TxHashArr = append(txs.TxHashArr, txHash)
|
||||
txs.VersionArr = append(txs.VersionArr, srcTx.Version)
|
||||
txs.LocktimeArr = append(txs.LocktimeArr, int64(srcTx.LockTime))
|
||||
txs.BlockHeightArr = append(txs.BlockHeightArr, int32(block.Header.Height))
|
||||
txs.BlockHashArr = append(txs.BlockHashArr, blockHash)
|
||||
txs.IdxArr = append(txs.IdxArr, int32(txIdx))
|
||||
|
||||
// Batch insert txins
|
||||
for idx, txin := range srcTx.TxIn {
|
||||
var witness string
|
||||
if len(txin.Witness) > 0 {
|
||||
witness = btcutils.WitnessToString(txin.Witness)
|
||||
}
|
||||
txins.TxHashArr = append(txins.TxHashArr, txHash)
|
||||
txins.TxIdxArr = append(txins.TxIdxArr, int32(idx))
|
||||
txins.PrevoutTxHashArr = append(txins.PrevoutTxHashArr, txin.PreviousOutTxHash.String())
|
||||
txins.PrevoutTxIdxArr = append(txins.PrevoutTxIdxArr, int32(txin.PreviousOutIndex))
|
||||
txins.ScriptsigArr = append(txins.ScriptsigArr, hex.EncodeToString(txin.SignatureScript))
|
||||
txins.WitnessArr = append(txins.WitnessArr, witness)
|
||||
txins.SequenceArr = append(txins.SequenceArr, int32(txin.Sequence))
|
||||
}
|
||||
|
||||
// Batch insert txouts
|
||||
for idx, txout := range srcTx.TxOut {
|
||||
txouts.TxHashArr = append(txouts.TxHashArr, txHash)
|
||||
txouts.TxIdxArr = append(txouts.TxIdxArr, int32(idx))
|
||||
txouts.PkscriptArr = append(txouts.PkscriptArr, hex.EncodeToString(txout.PkScript))
|
||||
txouts.ValueArr = append(txouts.ValueArr, txout.Value)
|
||||
}
|
||||
}
|
||||
}
|
||||
return blocks, txs, txouts, txins
|
||||
}
|
||||
|
||||
func mapTransactionModelToType(src gen.BitcoinTransaction, txInModel []gen.BitcoinTransactionTxin, txOutModels []gen.BitcoinTransactionTxout) (types.Transaction, error) {
|
||||
blockHash, err := chainhash.NewHashFromStr(src.BlockHash)
|
||||
if err != nil {
|
||||
return types.Transaction{}, errors.Wrap(err, "failed to parse block hash")
|
||||
}
|
||||
|
||||
txHash, err := chainhash.NewHashFromStr(src.TxHash)
|
||||
if err != nil {
|
||||
return types.Transaction{}, errors.Wrap(err, "failed to parse tx hash")
|
||||
}
|
||||
|
||||
// Sort txins and txouts by index (Asc)
|
||||
slices.SortFunc(txOutModels, func(i, j gen.BitcoinTransactionTxout) int {
|
||||
return cmp.Compare(i.TxIdx, j.TxIdx)
|
||||
})
|
||||
slices.SortFunc(txInModel, func(i, j gen.BitcoinTransactionTxin) int {
|
||||
return cmp.Compare(i.TxIdx, j.TxIdx)
|
||||
})
|
||||
|
||||
txIns := make([]*types.TxIn, 0, len(txInModel))
|
||||
txOuts := make([]*types.TxOut, 0, len(txOutModels))
|
||||
for _, txInModel := range txInModel {
|
||||
scriptsig, err := hex.DecodeString(txInModel.Scriptsig)
|
||||
if err != nil {
|
||||
return types.Transaction{}, errors.Wrap(err, "failed to decode scriptsig")
|
||||
}
|
||||
|
||||
prevoutTxHash, err := chainhash.NewHashFromStr(txInModel.PrevoutTxHash)
|
||||
if err != nil {
|
||||
return types.Transaction{}, errors.Wrap(err, "failed to parse prevout tx hash")
|
||||
}
|
||||
|
||||
witness, err := btcutils.WitnessFromString(txInModel.Witness)
|
||||
if err != nil {
|
||||
return types.Transaction{}, errors.Wrap(err, "failed to parse witness from hex string")
|
||||
}
|
||||
|
||||
txIns = append(txIns, &types.TxIn{
|
||||
SignatureScript: scriptsig,
|
||||
Witness: witness,
|
||||
Sequence: uint32(txInModel.Sequence),
|
||||
PreviousOutIndex: uint32(txInModel.PrevoutTxIdx),
|
||||
PreviousOutTxHash: *prevoutTxHash,
|
||||
})
|
||||
}
|
||||
for _, txOutModel := range txOutModels {
|
||||
pkscript, err := hex.DecodeString(txOutModel.Pkscript)
|
||||
if err != nil {
|
||||
return types.Transaction{}, errors.Wrap(err, "failed to decode pkscript")
|
||||
}
|
||||
txOuts = append(txOuts, &types.TxOut{
|
||||
PkScript: pkscript,
|
||||
Value: txOutModel.Value,
|
||||
})
|
||||
}
|
||||
|
||||
return types.Transaction{
|
||||
BlockHeight: int64(src.BlockHeight),
|
||||
BlockHash: *blockHash,
|
||||
Index: uint32(src.Idx),
|
||||
TxHash: *txHash,
|
||||
Version: src.Version,
|
||||
LockTime: uint32(src.Locktime),
|
||||
TxIn: txIns,
|
||||
TxOut: txOuts,
|
||||
}, nil
|
||||
}
|
||||
@@ -1,22 +0,0 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"github.com/gaze-network/indexer-network/internal/postgres"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/datagateway"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/repository/postgres/gen"
|
||||
)
|
||||
|
||||
// Make sure Repository implements the BitcoinDataGateway interface
|
||||
var _ datagateway.BitcoinDataGateway = (*Repository)(nil)
|
||||
|
||||
type Repository struct {
|
||||
db postgres.DB
|
||||
queries *gen.Queries
|
||||
}
|
||||
|
||||
func NewRepository(db postgres.DB) *Repository {
|
||||
return &Repository{
|
||||
db: db,
|
||||
queries: gen.New(db),
|
||||
}
|
||||
}
|
||||
@@ -1,35 +0,0 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
func (r *Repository) GetTransactionByHash(ctx context.Context, txHash chainhash.Hash) (*types.Transaction, error) {
|
||||
model, err := r.queries.GetTransactionByHash(ctx, txHash.String())
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, errors.Join(errs.NotFound, err)
|
||||
}
|
||||
return nil, errors.Wrap(err, "failed to get transaction by hash")
|
||||
}
|
||||
txIns, err := r.queries.GetTransactionTxInsByTxHashes(ctx, []string{txHash.String()})
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to get transaction txins by tx hashes")
|
||||
}
|
||||
txOuts, err := r.queries.GetTransactionTxOutsByTxHashes(ctx, []string{txHash.String()})
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, errors.Wrap(err, "failed to get transaction txouts by tx hashes")
|
||||
}
|
||||
|
||||
tx, err := mapTransactionModelToType(model, txIns, txOuts)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to map transaction model to type")
|
||||
}
|
||||
return &tx, nil
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
package bitcoin
|
||||
@@ -124,6 +124,7 @@ func (h *HttpHandler) GetTokenInfo(ctx *fiber.Ctx) (err error) {
|
||||
}
|
||||
circulatingSupply := mintedAmount.Sub(runeEntry.BurnedAmount)
|
||||
|
||||
terms := lo.FromPtr(runeEntry.Terms)
|
||||
resp := getTokenInfoResponse{
|
||||
Result: &getTokenInfoResult{
|
||||
Id: runeId,
|
||||
@@ -147,12 +148,12 @@ func (h *HttpHandler) GetTokenInfo(ctx *fiber.Ctx) (err error) {
|
||||
Spacers: runeEntry.SpacedRune.Spacers,
|
||||
Symbol: string(runeEntry.Symbol),
|
||||
Terms: entryTerms{
|
||||
Amount: lo.FromPtr(runeEntry.Terms.Amount),
|
||||
Cap: lo.FromPtr(runeEntry.Terms.Cap),
|
||||
HeightStart: runeEntry.Terms.HeightStart,
|
||||
HeightEnd: runeEntry.Terms.HeightEnd,
|
||||
OffsetStart: runeEntry.Terms.OffsetStart,
|
||||
OffsetEnd: runeEntry.Terms.OffsetEnd,
|
||||
Amount: lo.FromPtr(terms.Amount),
|
||||
Cap: lo.FromPtr(terms.Cap),
|
||||
HeightStart: terms.HeightStart,
|
||||
HeightEnd: terms.HeightEnd,
|
||||
OffsetStart: terms.OffsetStart,
|
||||
OffsetEnd: terms.OffsetEnd,
|
||||
},
|
||||
Turbo: runeEntry.Turbo,
|
||||
},
|
||||
|
||||
@@ -3,7 +3,7 @@ package config
|
||||
import "github.com/gaze-network/indexer-network/internal/postgres"
|
||||
|
||||
type Config struct {
|
||||
Datasource string `mapstructure:"datasource"` // Datasource to fetch bitcoin data for Meta-Protocol e.g. `bitcoin-node` | `database`
|
||||
Datasource string `mapstructure:"datasource"` // Datasource to fetch bitcoin data for Meta-Protocol e.g. `bitcoin-node`
|
||||
Database string `mapstructure:"database"` // Database to store runes data.
|
||||
APIHandlers []string `mapstructure:"api_handlers"` // List of API handlers to enable. (e.g. `http`)
|
||||
Postgres postgres.Config `mapstructure:"postgres"`
|
||||
|
||||
@@ -9,12 +9,12 @@ import (
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/indexers"
|
||||
"github.com/gaze-network/indexer-network/core/indexer"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/btcclient"
|
||||
"github.com/gaze-network/indexer-network/modules/runes/datagateway"
|
||||
"github.com/gaze-network/indexer-network/modules/runes/internal/entity"
|
||||
"github.com/gaze-network/indexer-network/modules/runes/runes"
|
||||
"github.com/gaze-network/indexer-network/pkg/btcclient"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger/slogx"
|
||||
"github.com/gaze-network/indexer-network/pkg/reportingclient"
|
||||
@@ -22,15 +22,15 @@ import (
|
||||
"github.com/samber/lo"
|
||||
)
|
||||
|
||||
var _ indexers.BitcoinProcessor = (*Processor)(nil)
|
||||
// Make sure to implement the Bitcoin Processor interface
|
||||
var _ indexer.Processor[*types.Block] = (*Processor)(nil)
|
||||
|
||||
type Processor struct {
|
||||
runesDg datagateway.RunesDataGateway
|
||||
indexerInfoDg datagateway.IndexerInfoDataGateway
|
||||
bitcoinClient btcclient.Contract
|
||||
bitcoinDataSource indexers.BitcoinDatasource
|
||||
network common.Network
|
||||
reportingClient *reportingclient.ReportingClient
|
||||
runesDg datagateway.RunesDataGateway
|
||||
indexerInfoDg datagateway.IndexerInfoDataGateway
|
||||
bitcoinClient btcclient.Contract
|
||||
network common.Network
|
||||
reportingClient *reportingclient.ReportingClient
|
||||
|
||||
newRuneEntries map[runes.RuneId]*runes.RuneEntry
|
||||
newRuneEntryStates map[runes.RuneId]*runes.RuneEntry
|
||||
@@ -40,12 +40,11 @@ type Processor struct {
|
||||
newRuneTxs []*entity.RuneTransaction
|
||||
}
|
||||
|
||||
func NewProcessor(runesDg datagateway.RunesDataGateway, indexerInfoDg datagateway.IndexerInfoDataGateway, bitcoinClient btcclient.Contract, bitcoinDataSource indexers.BitcoinDatasource, network common.Network, reportingClient *reportingclient.ReportingClient) *Processor {
|
||||
func NewProcessor(runesDg datagateway.RunesDataGateway, indexerInfoDg datagateway.IndexerInfoDataGateway, bitcoinClient btcclient.Contract, network common.Network, reportingClient *reportingclient.ReportingClient) *Processor {
|
||||
return &Processor{
|
||||
runesDg: runesDg,
|
||||
indexerInfoDg: indexerInfoDg,
|
||||
bitcoinClient: bitcoinClient,
|
||||
bitcoinDataSource: bitcoinDataSource,
|
||||
network: network,
|
||||
reportingClient: reportingClient,
|
||||
newRuneEntries: make(map[runes.RuneId]*runes.RuneEntry),
|
||||
@@ -98,7 +97,6 @@ func (p *Processor) ensureValidState(ctx context.Context) error {
|
||||
return errors.Wrapf(errs.ConflictSetting, "db version mismatch: current version is %d. Please upgrade to version %d", indexerState.DBVersion, DBVersion)
|
||||
}
|
||||
if indexerState.EventHashVersion != EventHashVersion {
|
||||
// TODO: automate reset db instead of returning error
|
||||
return errors.Wrapf(errs.ConflictSetting, "event version mismatch: current version is %d. Please reset rune's db first.", indexerState.EventHashVersion, EventHashVersion)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/modules/runes/datagateway"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/jackc/pgx"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
var ErrTxAlreadyExists = errors.New("Transaction already exists. Call Commit() or Rollback() first.")
|
||||
|
||||
89
modules/runes/runes.go
Normal file
89
modules/runes/runes.go
Normal file
@@ -0,0 +1,89 @@
|
||||
package runes
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/btcsuite/btcd/rpcclient"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/core/datasources"
|
||||
"github.com/gaze-network/indexer-network/core/indexer"
|
||||
"github.com/gaze-network/indexer-network/core/types"
|
||||
"github.com/gaze-network/indexer-network/internal/config"
|
||||
"github.com/gaze-network/indexer-network/internal/postgres"
|
||||
runesapi "github.com/gaze-network/indexer-network/modules/runes/api"
|
||||
runesdatagateway "github.com/gaze-network/indexer-network/modules/runes/datagateway"
|
||||
runespostgres "github.com/gaze-network/indexer-network/modules/runes/repository/postgres"
|
||||
runesusecase "github.com/gaze-network/indexer-network/modules/runes/usecase"
|
||||
"github.com/gaze-network/indexer-network/pkg/btcclient"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/reportingclient"
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/samber/do/v2"
|
||||
"github.com/samber/lo"
|
||||
)
|
||||
|
||||
func New(injector do.Injector) (indexer.IndexerWorker, error) {
|
||||
ctx := do.MustInvoke[context.Context](injector)
|
||||
conf := do.MustInvoke[config.Config](injector)
|
||||
reportingClient := do.MustInvoke[*reportingclient.ReportingClient](injector)
|
||||
|
||||
var (
|
||||
runesDg runesdatagateway.RunesDataGateway
|
||||
indexerInfoDg runesdatagateway.IndexerInfoDataGateway
|
||||
)
|
||||
switch strings.ToLower(conf.Modules.Runes.Database) {
|
||||
case "postgresql", "postgres", "pg":
|
||||
pg, err := postgres.NewPool(ctx, conf.Modules.Runes.Postgres)
|
||||
if err != nil {
|
||||
if errors.Is(err, errs.InvalidArgument) {
|
||||
return nil, errors.Wrap(err, "Invalid Postgres configuration for indexer")
|
||||
}
|
||||
return nil, errors.Wrap(err, "can't create Postgres connection pool")
|
||||
}
|
||||
defer pg.Close()
|
||||
runesRepo := runespostgres.NewRepository(pg)
|
||||
runesDg = runesRepo
|
||||
indexerInfoDg = runesRepo
|
||||
default:
|
||||
return nil, errors.Wrapf(errs.Unsupported, "%q database for indexer is not supported", conf.Modules.Runes.Database)
|
||||
}
|
||||
|
||||
var bitcoinDatasource datasources.Datasource[*types.Block]
|
||||
var bitcoinClient btcclient.Contract
|
||||
switch strings.ToLower(conf.Modules.Runes.Datasource) {
|
||||
case "bitcoin-node":
|
||||
btcClient := do.MustInvoke[*rpcclient.Client](injector)
|
||||
bitcoinNodeDatasource := datasources.NewBitcoinNode(btcClient)
|
||||
bitcoinDatasource = bitcoinNodeDatasource
|
||||
bitcoinClient = bitcoinNodeDatasource
|
||||
default:
|
||||
return nil, errors.Wrapf(errs.Unsupported, "%q datasource is not supported", conf.Modules.Runes.Datasource)
|
||||
}
|
||||
|
||||
processor := NewProcessor(runesDg, indexerInfoDg, bitcoinClient, conf.Network, reportingClient)
|
||||
if err := processor.VerifyStates(ctx); err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
}
|
||||
|
||||
// Mount API
|
||||
apiHandlers := lo.Uniq(conf.Modules.Runes.APIHandlers)
|
||||
for _, handler := range apiHandlers {
|
||||
switch handler { // TODO: support more handlers (e.g. gRPC)
|
||||
case "http":
|
||||
httpServer := do.MustInvoke[*fiber.App](injector)
|
||||
runesUsecase := runesusecase.New(runesDg, bitcoinClient)
|
||||
runesHTTPHandler := runesapi.NewHTTPHandler(conf.Network, runesUsecase)
|
||||
if err := runesHTTPHandler.Mount(httpServer); err != nil {
|
||||
return nil, errors.Wrap(err, "can't mount Runes API")
|
||||
}
|
||||
logger.InfoContext(ctx, "Mounted HTTP handler")
|
||||
default:
|
||||
return nil, errors.Wrapf(errs.Unsupported, "%q API handler is not supported", handler)
|
||||
}
|
||||
}
|
||||
|
||||
indexer := indexer.New(processor, bitcoinDatasource)
|
||||
return indexer, nil
|
||||
}
|
||||
@@ -117,7 +117,8 @@ func (e RuneEntry) Supply() (uint128.Uint128, error) {
|
||||
}
|
||||
|
||||
func (e RuneEntry) MintedAmount() (uint128.Uint128, error) {
|
||||
amount, overflow := e.Mints.MulOverflow(lo.FromPtr(e.Terms.Amount))
|
||||
terms := lo.FromPtr(e.Terms)
|
||||
amount, overflow := e.Mints.MulOverflow(lo.FromPtr(terms.Amount))
|
||||
if overflow {
|
||||
return uint128.Uint128{}, errors.WithStack(errs.OverflowUint128)
|
||||
}
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
package usecase
|
||||
|
||||
import (
|
||||
"github.com/gaze-network/indexer-network/modules/bitcoin/btcclient"
|
||||
"github.com/gaze-network/indexer-network/modules/runes/datagateway"
|
||||
"github.com/gaze-network/indexer-network/pkg/btcclient"
|
||||
)
|
||||
|
||||
type Usecase struct {
|
||||
|
||||
99
pkg/automaxprocs/automaxprocs.go
Normal file
99
pkg/automaxprocs/automaxprocs.go
Normal file
@@ -0,0 +1,99 @@
|
||||
package automaxprocs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"runtime"
|
||||
|
||||
"github.com/Cleverse/go-utilities/utils"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger/slogx"
|
||||
"go.uber.org/automaxprocs/maxprocs"
|
||||
)
|
||||
|
||||
var (
|
||||
// undo is the undo function returned by maxprocs.Set
|
||||
undo func()
|
||||
|
||||
// autoMaxProcs is the value of GOMAXPROCS set by automaxprocs.
|
||||
// will be -1 if `automaxprocs` is not initialized.
|
||||
autoMaxProcs = -1
|
||||
|
||||
// initialMaxProcs is the initial value of GOMAXPROCS.
|
||||
initialMaxProcs = Current()
|
||||
)
|
||||
|
||||
func Init() error {
|
||||
logger := logger.With(
|
||||
slogx.String("package", "automaxprocs"),
|
||||
slogx.String("event", "set_gomaxprocs"),
|
||||
slogx.Int("prev_maxprocs", initialMaxProcs),
|
||||
)
|
||||
|
||||
// Create a logger function for `maxprocs.Set`.
|
||||
setMaxProcLogger := func(format string, v ...any) {
|
||||
fields := make([]slog.Attr, 0, 1)
|
||||
|
||||
// `maxprocs.Set` will always pass current GOMAXPROCS value to logger.
|
||||
// except when calling `undo` function, it will not pass any value.
|
||||
if val, ok := utils.Optional(v); ok {
|
||||
// if `GOMAXPROCS` environment variable is set, then `automaxprocs` will honor it.
|
||||
if _, exists := os.LookupEnv("GOMAXPROCS"); exists {
|
||||
val = Current()
|
||||
}
|
||||
|
||||
// add logging field for `set_maxprocs` value if it's present in integer value.
|
||||
if setmaxprocs, ok := val.(int); ok {
|
||||
fields = append(fields, slogx.Int("set_maxprocs", setmaxprocs))
|
||||
}
|
||||
}
|
||||
|
||||
logger.LogAttrs(context.Background(), slog.LevelInfo, fmt.Sprintf(format, v...), fields...)
|
||||
}
|
||||
|
||||
// Set GOMAXPROCS to match the Linux container CPU quota (if any), returning
|
||||
// any error encountered and an undo function.
|
||||
//
|
||||
// Set is a no-op on non-Linux systems and in Linux environments without a
|
||||
// configured CPU quota.
|
||||
revert, err := maxprocs.Set(maxprocs.Logger(setMaxProcLogger), maxprocs.Min(1))
|
||||
if err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
|
||||
// set the result of `maxprocs.Set` to global variable.
|
||||
autoMaxProcs = Current()
|
||||
undo = revert
|
||||
return nil
|
||||
}
|
||||
|
||||
// Undo restores GOMAXPROCS to its previous value.
|
||||
// or revert to initial value if `automaxprocs` is not initialized.
|
||||
//
|
||||
// returns the current GOMAXPROCS value.
|
||||
func Undo() int {
|
||||
if undo != nil {
|
||||
undo()
|
||||
return Current()
|
||||
}
|
||||
|
||||
runtime.GOMAXPROCS(initialMaxProcs)
|
||||
return initialMaxProcs
|
||||
}
|
||||
|
||||
// Current returns the current value of GOMAXPROCS.
|
||||
func Current() int {
|
||||
return runtime.GOMAXPROCS(0)
|
||||
}
|
||||
|
||||
// Value returns the value of GOMAXPROCS set by automaxprocs.
|
||||
// returns -1 if `automaxprocs` is not initialized.
|
||||
func Value() int {
|
||||
if autoMaxProcs <= 0 {
|
||||
return -1
|
||||
}
|
||||
return autoMaxProcs
|
||||
}
|
||||
21
pkg/btcutils/satoshi.go
Normal file
21
pkg/btcutils/satoshi.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package btcutils
|
||||
|
||||
import "github.com/shopspring/decimal"
|
||||
|
||||
const (
|
||||
BitcoinDecimals = 8
|
||||
)
|
||||
|
||||
// satsUnit is 10^8
|
||||
var satsUnit = decimal.New(1, BitcoinDecimals)
|
||||
|
||||
// BitcoinToSatoshi converts a amount in Bitcoin format to Satoshi format.
|
||||
func BitcoinToSatoshi(v float64) int64 {
|
||||
amount := decimal.NewFromFloat(v)
|
||||
return amount.Mul(satsUnit).IntPart()
|
||||
}
|
||||
|
||||
// SatoshiToBitcoin converts a amount in Satoshi format to Bitcoin format.
|
||||
func SatoshiToBitcoin(v int64) float64 {
|
||||
return decimal.New(v, -BitcoinDecimals).InexactFloat64()
|
||||
}
|
||||
39
pkg/btcutils/satoshi_test.go
Normal file
39
pkg/btcutils/satoshi_test.go
Normal file
@@ -0,0 +1,39 @@
|
||||
package btcutils
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestSatoshiConversion(t *testing.T) {
|
||||
testcases := []struct {
|
||||
sats float64
|
||||
btc int64
|
||||
}{
|
||||
{2.29980951, 229980951},
|
||||
{1.29609085, 129609085},
|
||||
{1.2768897, 127688970},
|
||||
{0.62518296, 62518296},
|
||||
{0.29998462, 29998462},
|
||||
{0.1251, 12510000},
|
||||
{0.02016011, 2016011},
|
||||
{0.0198473, 1984730},
|
||||
{0.0051711, 517110},
|
||||
{0.0012, 120000},
|
||||
{7e-05, 7000},
|
||||
{3.835e-05, 3835},
|
||||
{1.962e-05, 1962},
|
||||
}
|
||||
for _, testcase := range testcases {
|
||||
t.Run(fmt.Sprintf("BtcToSats/%v", testcase.sats), func(t *testing.T) {
|
||||
require.NotEqual(t, testcase.btc, int64(testcase.sats*1e8), "Testcase value should have precision error")
|
||||
assert.Equal(t, testcase.btc, BitcoinToSatoshi(testcase.sats))
|
||||
})
|
||||
t.Run(fmt.Sprintf("SatsToBtc/%v", testcase.sats), func(t *testing.T) {
|
||||
assert.Equal(t, testcase.sats, SatoshiToBitcoin(testcase.btc))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/hex"
|
||||
"strings"
|
||||
|
||||
"github.com/Cleverse/go-utilities/utils"
|
||||
"github.com/btcsuite/btcd/wire"
|
||||
"github.com/cockroachdb/errors"
|
||||
)
|
||||
@@ -12,6 +13,9 @@ const (
|
||||
witnessSeparator = " "
|
||||
)
|
||||
|
||||
// CoinbaseWitness is the witness data for a coinbase transaction.
|
||||
var CoinbaseWitness = utils.Must(WitnessFromHex([]string{"0000000000000000000000000000000000000000000000000000000000000000"}))
|
||||
|
||||
// WitnessToHex formats the passed witness stack as a slice of hex-encoded strings.
|
||||
func WitnessToHex(witness wire.TxWitness) []string {
|
||||
if len(witness) == 0 {
|
||||
@@ -37,7 +41,10 @@ func WitnessToString(witness wire.TxWitness) string {
|
||||
|
||||
// WitnessFromHex parses the passed slice of hex-encoded strings into a witness stack.
|
||||
func WitnessFromHex(witnesses []string) (wire.TxWitness, error) {
|
||||
// NOTE: some witness from bitcoin node are empty and some are nil(most are nil), it's not clear why.
|
||||
// For now, we will return nil for both cases.
|
||||
if len(witnesses) == 0 {
|
||||
// return wire.TxWitness{}, nil
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
"github.com/gaze-network/indexer-network/pkg/stacktrace"
|
||||
)
|
||||
|
||||
func middlewareError() middleware {
|
||||
func middlewareErrorStackTrace() middleware {
|
||||
return func(next handleFunc) handleFunc {
|
||||
return func(ctx context.Context, rec slog.Record) error {
|
||||
rec.Attrs(func(attr slog.Attr) bool {
|
||||
|
||||
@@ -133,9 +133,7 @@ var (
|
||||
}
|
||||
|
||||
// Default Middlewares
|
||||
defaultMiddleware = []middleware{
|
||||
middlewareError(),
|
||||
}
|
||||
defaultMiddleware = []middleware{}
|
||||
)
|
||||
|
||||
// Init initializes global logger and slog logger with given configuration.
|
||||
@@ -147,12 +145,14 @@ func Init(cfg Config) error {
|
||||
Level: lvl,
|
||||
ReplaceAttr: attrReplacerChain(defaultAttrReplacers...),
|
||||
}
|
||||
middlewares = append([]middleware{}, defaultMiddleware...)
|
||||
)
|
||||
|
||||
lvl.Set(slog.LevelInfo)
|
||||
if cfg.Debug {
|
||||
lvl.Set(slog.LevelDebug)
|
||||
options.AddSource = true
|
||||
middlewares = append(middlewares, middlewareErrorStackTrace())
|
||||
}
|
||||
|
||||
switch strings.ToLower(cfg.Output) {
|
||||
@@ -165,7 +165,7 @@ func Init(cfg Config) error {
|
||||
handler = slog.NewTextHandler(os.Stdout, options)
|
||||
}
|
||||
|
||||
logger = slog.New(newChainHandlers(handler, defaultMiddleware...))
|
||||
logger = slog.New(newChainHandlers(handler, middlewares...))
|
||||
slog.SetDefault(logger)
|
||||
return nil
|
||||
}
|
||||
|
||||
133
pkg/parquetutils/buffer.go
Normal file
133
pkg/parquetutils/buffer.go
Normal file
@@ -0,0 +1,133 @@
|
||||
// nolint: wrapcheck
|
||||
package parquetutils
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"io"
|
||||
"sync"
|
||||
|
||||
"github.com/xitongsys/parquet-go/source"
|
||||
)
|
||||
|
||||
var (
|
||||
// Make sure Buffer implements the ParquetFile interface.
|
||||
_ source.ParquetFile = (*Buffer)(nil)
|
||||
|
||||
// Make sure Buffer implements the io.WriterAt interface.
|
||||
_ io.WriterAt = (*Buffer)(nil)
|
||||
)
|
||||
|
||||
// Buffer allows reading parquet messages from a memory buffer.
|
||||
type Buffer struct {
|
||||
buf []byte
|
||||
loc int
|
||||
m sync.Mutex
|
||||
}
|
||||
|
||||
// NewBuffer creates a new in memory parquet buffer.
|
||||
func NewBuffer() *Buffer {
|
||||
return &Buffer{buf: make([]byte, 0, 512)}
|
||||
}
|
||||
|
||||
// NewBufferFrom creates new in memory parquet buffer from the given bytes.
|
||||
// It uses the provided slice as its buffer.
|
||||
func NewBufferFrom(s []byte) *Buffer {
|
||||
return &Buffer{
|
||||
buf: s,
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Buffer) Create(string) (source.ParquetFile, error) {
|
||||
return &Buffer{buf: make([]byte, 0, 512)}, nil
|
||||
}
|
||||
|
||||
func (b *Buffer) Open(string) (source.ParquetFile, error) {
|
||||
return NewBufferFrom(b.Bytes()), nil
|
||||
}
|
||||
|
||||
// Seek seeks in the underlying memory buffer.
|
||||
func (b *Buffer) Seek(offset int64, whence int) (int64, error) {
|
||||
newLoc := b.loc
|
||||
switch whence {
|
||||
case io.SeekStart:
|
||||
newLoc = int(offset)
|
||||
case io.SeekCurrent:
|
||||
newLoc += int(offset)
|
||||
case io.SeekEnd:
|
||||
newLoc = len(b.buf) + int(offset)
|
||||
default:
|
||||
return int64(b.loc), errors.New("Seek: invalid whence")
|
||||
}
|
||||
|
||||
if newLoc < 0 {
|
||||
return int64(b.loc), errors.New("Seek: invalid offset")
|
||||
}
|
||||
|
||||
if newLoc > len(b.buf) {
|
||||
newLoc = len(b.buf)
|
||||
}
|
||||
|
||||
b.loc = newLoc
|
||||
return int64(b.loc), nil
|
||||
}
|
||||
|
||||
// Read reads data form BufferFile into p.
|
||||
func (b *Buffer) Read(p []byte) (n int, err error) {
|
||||
n = copy(p, b.buf[b.loc:len(b.buf)])
|
||||
b.loc += n
|
||||
|
||||
if b.loc == len(b.buf) {
|
||||
return n, io.EOF
|
||||
}
|
||||
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// Write writes data from p into BufferFile.
|
||||
func (b *Buffer) Write(p []byte) (n int, err error) {
|
||||
n, err = b.WriteAt(p, int64(b.loc))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
b.loc += n
|
||||
return
|
||||
}
|
||||
|
||||
// WriteAt writes a slice of bytes to a buffer starting at the position provided
|
||||
// The number of bytes written will be returned, or error. Can overwrite previous
|
||||
// written slices if the write ats overlap.
|
||||
func (b *Buffer) WriteAt(p []byte, pos int64) (n int, err error) {
|
||||
b.m.Lock()
|
||||
defer b.m.Unlock()
|
||||
pLen := len(p)
|
||||
expLen := pos + int64(pLen)
|
||||
if int64(len(b.buf)) < expLen {
|
||||
if int64(cap(b.buf)) < expLen {
|
||||
newBuf := make([]byte, expLen)
|
||||
copy(newBuf, b.buf)
|
||||
b.buf = newBuf
|
||||
}
|
||||
b.buf = b.buf[:expLen]
|
||||
}
|
||||
copy(b.buf[pos:], p)
|
||||
return pLen, nil
|
||||
}
|
||||
|
||||
// Close is a no-op for a memory buffer.
|
||||
func (*Buffer) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Bytes returns the underlying buffer bytes.
|
||||
func (b *Buffer) Bytes() []byte {
|
||||
return b.buf
|
||||
}
|
||||
|
||||
// Reset resets the buffer to be empty,
|
||||
// but it retains the underlying storage for use by future writes.
|
||||
func (b *Buffer) Reset() {
|
||||
b.m.Lock()
|
||||
defer b.m.Unlock()
|
||||
b.buf = b.buf[:0]
|
||||
b.loc = 0
|
||||
}
|
||||
26
pkg/parquetutils/parquetutils.go
Normal file
26
pkg/parquetutils/parquetutils.go
Normal file
@@ -0,0 +1,26 @@
|
||||
package parquetutils
|
||||
|
||||
import (
|
||||
"github.com/pkg/errors"
|
||||
"github.com/xitongsys/parquet-go/reader"
|
||||
"github.com/xitongsys/parquet-go/source"
|
||||
)
|
||||
|
||||
// ReaderConcurrency parallel number of file readers.
|
||||
var ReaderConcurrency int64 = 8
|
||||
|
||||
// ReadAll reads all records from the parquet file.
|
||||
func ReadAll[T any](sourceFile source.ParquetFile) ([]T, error) {
|
||||
r, err := reader.NewParquetReader(sourceFile, new(T), ReaderConcurrency)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "can't create parquet reader")
|
||||
}
|
||||
defer r.ReadStop()
|
||||
|
||||
data := make([]T, r.GetNumRows())
|
||||
if err = r.Read(&data); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to read parquet data")
|
||||
}
|
||||
|
||||
return data, nil
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/cockroachdb/errors"
|
||||
"github.com/gaze-network/indexer-network/common"
|
||||
"github.com/gaze-network/indexer-network/common/errs"
|
||||
"github.com/gaze-network/indexer-network/pkg/httpclient"
|
||||
"github.com/gaze-network/indexer-network/pkg/logger"
|
||||
)
|
||||
@@ -35,7 +36,7 @@ func New(config Config) (*ReportingClient, error) {
|
||||
return nil, errors.Wrap(err, "can't create http client")
|
||||
}
|
||||
if config.Name == "" {
|
||||
return nil, errors.New("reporting.name config is required if reporting is enabled")
|
||||
return nil, errors.Wrap(errs.InvalidArgument, "reporting.name config is required if reporting is enabled")
|
||||
}
|
||||
return &ReportingClient{
|
||||
httpClient: httpClient,
|
||||
|
||||
10
sqlc.yaml
10
sqlc.yaml
@@ -7,16 +7,6 @@
|
||||
# https://docs.sqlc.dev/en/stable/howto/ddl.html#golang-migrate
|
||||
version: "2"
|
||||
sql:
|
||||
- schema: "./modules/bitcoin/database/postgresql/migrations"
|
||||
queries: "./modules/bitcoin/database/postgresql/queries"
|
||||
engine: "postgresql"
|
||||
gen:
|
||||
go:
|
||||
package: "gen"
|
||||
out: "./modules/bitcoin/repository/postgres/gen"
|
||||
sql_package: "pgx/v5"
|
||||
rename:
|
||||
id: "Id"
|
||||
- schema: "./modules/runes/database/postgresql/migrations"
|
||||
queries: "./modules/runes/database/postgresql/queries"
|
||||
engine: "postgresql"
|
||||
|
||||
Reference in New Issue
Block a user